KAFKA-17593; [7/N] Introduce CoordinatorExecutor #17823
Open
+737
−4
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
Please enter the commit message for your changes. Lines starting
dajac
commented
Nov 15, 2024
| @@ -642,6 +664,7 @@ public void testScheduleLoading() { | |||
| when(builder.withTimer(any())).thenReturn(builder); | |||
| when(builder.withCoordinatorMetrics(any())).thenReturn(builder); | |||
| when(builder.withTopicPartition(any())).thenReturn(builder); | |||
| when(builder.withExecutor(any())).thenReturn(builder); | |||
There was a problem hiding this comment.
fyi, i'll refactor those tests in a separate PR. i'll also so some cleanups.
|
@smjn fyi. Potentially this is useful for the record deletion work that you're doing. |
lianetm
reviewed
Nov 15, 2024
Comment on lines
+80
to
+83
| if (tasks.containsKey(key)) return false; | ||
|
|
||
| // We use the task as a lock for the key. | ||
| tasks.put(key, task); |
There was a problem hiding this comment.
what about
Suggested change
| if (tasks.containsKey(key)) return false; | |
| // We use the task as a lock for the key. | |
| tasks.put(key, task); | |
| if (tasks.putIfAbsent(key, task) != null) return false; |
| // that the task was either replaced or cancelled. We stop. | ||
| if (tasks.get(key) != task) return; | ||
|
|
||
| // Executor the task. |
There was a problem hiding this comment.
Suggested change
| // Executor the task. | |
| // Execute the task. |
| return operation.onComplete(result.result, result.exception); | ||
| } | ||
| ).exceptionally(exception -> { | ||
| if (exception instanceof NotCoordinatorException || exception instanceof CoordinatorLoadInProgressException) { |
There was a problem hiding this comment.
could we end up here due to a failure on scheduling the write op, without having removed the task? (ln 102). Wondering if we need to remove it here too to clean up.
| } | ||
|
|
||
| /** | ||
| * Schedule an asynchronous tasks. Note that only one task for a given key can |
There was a problem hiding this comment.
Suggested change
| * Schedule an asynchronous tasks. Note that only one task for a given key can | |
| * Schedule an asynchronous task. Note that only one task for a given key can |
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.
This patch introduces the
CoordinatorExecutorconstruct into theCoordinatorRuntime. It allows scheduling asynchronous tasks from within aCoordinatorShardwhile respecting the runtime semantic. It will be used to asynchronously resolve regular expressions.The
GroupCoordinatorServiceuses a defaultExecutorServicewith a single thread to back it at the moment. It seems that it should be sufficient. In the future, we could consider making the number of threads configurable.Committer Checklist (excluded from commit message)