Skip to content

Change the relationship between txn and task #703

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
morningman merged 2 commits into from
Mar 7, 2019
Merged

Change the relationship between txn and task #703

morningman merged 2 commits into from
Mar 7, 2019

Conversation

@EmmyMiao87
Copy link
Contributor

@EmmyMiao87 EmmyMiao87 commented Mar 7, 2019

  1. Check if properties is null before check routine load properties
  2. Change transactionStateChange reason to string
  3. calculate current num by beId
  4. Add kafka offset properties
  5. Perfer to use previous be id
  6. Add before commit listerner of txn: if txn is committed after task is aborted, commit will be aborted
  7. queryId of stream load plan = taskId

@EmmyMiao87
Change the relationskip between txn and task
08d3207 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Perfer to use previous be id
6. Add before commit listerner of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
@EmmyMiao87 EmmyMiao87 force-pushed the kafka_routine_load branch from 6c54104 to 08d3207 Compare March 7, 2019 10:48
morningman
morningman requested changes Mar 7, 2019
View reviewed changes
fe/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
kafkaOffsets = new ArrayList<>();
String[] kafkaOffsetsStringList = customProperties.get(KAFKA_OFFSETS_PROPERTY).split(",");
for (String s : kafkaOffsetsStringList) {
kafkaOffsets.add(Long.valueOf(s));
Copy link
Contributor

@morningman morningman Mar 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You should check the number format. And also check whether the number of specified offsets equals to number of specified partitions

Copy link
Contributor Author

@EmmyMiao87 EmmyMiao87 Mar 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will do it later

fe/src/main/java/org/apache/doris/load/routineload/KafkaProgress.java
KafkaProgress newProgress = (KafkaProgress) progress;
newProgress.getPartitionIdToOffset().entrySet().parallelStream()
.forEach(entity -> partitionIdToOffset.put(entity.getKey(), entity.getValue()));
.forEach(entity -> partitionIdToOffset.put(entity.getKey(), entity.getValue() + 1));
Copy link
Contributor

@morningman morningman Mar 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why + 1？

Copy link
Contributor Author

@EmmyMiao87 EmmyMiao87 Mar 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The offset of txn is the end offset while the kafka offset of progress is begin offset . The next task will read data from this begin offset.

fe/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java Outdated
}
String taskId = txnState.getLabel();
if (routineLoadTaskInfoList.parallelStream().anyMatch(entity -> entity.getId().toString().equals(taskId))) {
LOG.debug("there are a txn of routine load task will be aborted");
Copy link
Contributor

@morningman morningman Mar 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"there are a txn" ??

Copy link
Contributor

@morningman morningman Mar 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

And better to add taskid in log

@morningman morningman merged commit 19c1ff3 into apache:kafka_routine_load Mar 7, 2019
morningman pushed a commit to morningman/doris that referenced this pull request Mar 25, 2019
@EmmyMiao87 @morningman
Change the relationship between txn and task (apache#703)
09777e5 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
morningman pushed a commit that referenced this pull request Mar 25, 2019
@EmmyMiao87 @morningman
Change the relationship between txn and task (#703)
ac90ffb 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
morningman pushed a commit to morningman/doris that referenced this pull request Apr 4, 2019
@EmmyMiao87 @morningman
Change the relationship between txn and task (apache#703)
6a2f70a 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
morningman pushed a commit that referenced this pull request Apr 4, 2019
@EmmyMiao87 @morningman
Change the relationship between txn and task (#703)
29ac88d 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
morningman pushed a commit that referenced this pull request Apr 19, 2019
@EmmyMiao87 @morningman
Change the relationship between txn and task (#703)
0b25171 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
EmmyMiao87 added a commit to EmmyMiao87/incubator-doris that referenced this pull request Apr 23, 2019
@EmmyMiao87
Change the relationship between txn and task (apache#703)
fa84921 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
morningman pushed a commit that referenced this pull request Apr 24, 2019
@EmmyMiao87 @morningman
Change the relationship between txn and task (#703)
16c18e1 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
imay pushed a commit that referenced this pull request Apr 28, 2019
@EmmyMiao87 @imay
Change the relationship between txn and task (#703)
fbbe0d1 
1. Check if properties is null before check routine load properties
2. Change transactionStateChange reason to string
3. calculate current num by beId
4. Add kafka offset properties
5. Prefer to use previous be id
6. Add before commit listener of txn: if txn is committed after task is aborted, commit will be aborted
7. queryId of stream load plan = taskId
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@morningman morningman morningman requested changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@EmmyMiao87 @morningman