Concurrent replace should work with supervisors using concurrent locks #15995
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
kfaraz
reviewed
Feb 28, 2024
Comment on lines
90
to
91
| Boolean useConcurrentLocks = Tasks.DEFAULT_USE_CONCURRENT_LOCKS; | ||
| TaskLockType taskLockType = Tasks.DEFAULT_TASK_LOCK_TYPE; |
There was a problem hiding this comment.
Suggested change
| Boolean useConcurrentLocks = Tasks.DEFAULT_USE_CONCURRENT_LOCKS; | |
| TaskLockType taskLockType = Tasks.DEFAULT_TASK_LOCK_TYPE; | |
| boolean hasAppendLock = Tasks.DEFAULT_USE_CONCURRENT_LOCKS; |
| } | ||
| } | ||
|
|
||
| if (supervisor instanceof SeekableStreamSupervisor | ||
| && !supervisorSpec.isSuspended() | ||
| && supervisorSpec.getDataSources().contains(datasource) | ||
| && TaskLockType.APPEND.equals(taskLockType)) { | ||
| && (useConcurrentLocks || TaskLockType.APPEND.equals(taskLockType))) { |
There was a problem hiding this comment.
Suggested change
| && (useConcurrentLocks || TaskLockType.APPEND.equals(taskLockType))) { | |
| && hasAppendLock) { |
Comment on lines
96
to
108
| useConcurrentLocks = QueryContexts.getAsBoolean( | ||
| Tasks.USE_CONCURRENT_LOCKS, | ||
| context.get(Tasks.USE_CONCURRENT_LOCKS) | ||
| ); | ||
|
|
||
| if (useConcurrentLocks == null) { | ||
| useConcurrentLocks = false; | ||
| taskLockType = QueryContexts.getAsEnum( | ||
| Tasks.TASK_LOCK_TYPE, | ||
| context.get(Tasks.TASK_LOCK_TYPE), | ||
| TaskLockType.class | ||
| ); | ||
| } |
There was a problem hiding this comment.
Suggested change
| useConcurrentLocks = QueryContexts.getAsBoolean( | |
| Tasks.USE_CONCURRENT_LOCKS, | |
| context.get(Tasks.USE_CONCURRENT_LOCKS) | |
| ); | |
| if (useConcurrentLocks == null) { | |
| useConcurrentLocks = false; | |
| taskLockType = QueryContexts.getAsEnum( | |
| Tasks.TASK_LOCK_TYPE, | |
| context.get(Tasks.TASK_LOCK_TYPE), | |
| TaskLockType.class | |
| ); | |
| } | |
| Boolean useConcurrentLocks = QueryContexts.getAsBoolean( | |
| Tasks.USE_CONCURRENT_LOCKS, | |
| context.get(Tasks.USE_CONCURRENT_LOCKS) | |
| ); | |
| if (useConcurrentLocks == null) { | |
| TaskLockType taskLockType = QueryContexts.getAsEnum( | |
| Tasks.TASK_LOCK_TYPE, | |
| context.get(Tasks.TASK_LOCK_TYPE), | |
| TaskLockType.class | |
| ); | |
| if (taskLockType == null) { | |
| hasAppendLock = Tasks.DEFAULT_USE_CONCURRENT_LOCKS; | |
| } else if (taskLockType == TaskLockType.APPEND) { | |
| hasAppendLock = true; | |
| } else { | |
| hasAppendLock = false; | |
| } | |
| } else { | |
| hasAppendLock = useConcurrentLocks; | |
| } |
kfaraz
approved these changes
Feb 29, 2024
|
Thank you for the review @kfaraz |
317brian
pushed a commit
to 317brian/druid
that referenced
this pull request
Feb 29, 2024
apache#15995) * Concurrent replace should work with supervisors using concurrent locks * Ignore supervisors with useConcurrentLocks set to false * Apply feedback
AmatyaAvadhanula
added a commit
to AmatyaAvadhanula/druid
that referenced
this pull request
Mar 1, 2024
apache#15995) * Concurrent replace should work with supervisors using concurrent locks * Ignore supervisors with useConcurrentLocks set to false * Apply feedback
asdf2014
added a commit
that referenced
this pull request
Mar 4, 2024
* docs: add mermaid diagram support * fix crash when parsing data in data loader that can not be parsed (#15983) * update jetty to address CVE (#16000) * Concurrent replace should work with supervisors using concurrent locks (#15995) * Concurrent replace should work with supervisors using concurrent locks * Ignore supervisors with useConcurrentLocks set to false * Apply feedback * Add pre-check for heavy debug logs (#15706) Co-authored-by: Kashif Faraz <kashif.faraz@gmail.com> Co-authored-by: Benedict Jin <asdf2014@apache.org> * Remove helm paths from CodeQL config (#16006) * docs: mention acid-compliance for metadb --------- Co-authored-by: Vadim Ogievetsky <vadim@ogievetsky.com> Co-authored-by: Jan Werner <105367074+janjwerner-confluent@users.noreply.github.com> Co-authored-by: AmatyaAvadhanula <amatya.avadhanula@imply.io> Co-authored-by: Sensor <fectrain@outlook.com> Co-authored-by: Kashif Faraz <kashif.faraz@gmail.com> Co-authored-by: Benedict Jin <asdf2014@apache.org>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
A concurrent replace should allow the existing pending segments of a peon to be upgraded via the supervisor.
Currently, we check if there is such an active supervisor for streaming ingestion when the context value of
taskLockType:APPEND.This PR ensures that
useConcurrentLocks:trueworks as well.Please note that if
useConcurrentLocksis set tofalse, it would supersedetaskLockType:APPENDThis PR has: