bbejeck commented on code in PR #17795:
URL: https://github.com/apache/kafka/pull/17795#discussion_r1856742919
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java:
##########
@@ -281,17 +300,173 @@ public String toString() {
}
}
+ private final BlockingQueue<BackgroundEvent> onCallbackRequests = new
LinkedBlockingQueue<>();
+
+ private ApplicationEventHandler applicationEventHandler = null;
+
+ private Optional<Function<Set<StreamsAssignmentInterface.TaskId>,
Optional<Exception>>> onTasksRevokedCallback = null;
+ private Optional<Function<Assignment, Optional<Exception>>>
onTasksAssignedCallback = null;
+ private Optional<Supplier<Optional<Exception>>> onAllTasksLostCallback =
null;
+
+ private final StreamsRebalanceEventProcessor
streamsRebalanceEventProcessor;
+
+ private class StreamsRebalanceEventProcessor implements
EventProcessor<BackgroundEvent> {
+
+ @Override
+ public void process(final BackgroundEvent event) {
+ switch (event.type()) {
+ case ERROR:
+ process((ErrorEvent) event);
Review Comment:
Maybe just have `throw event.error();` here instead of the one-line method.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java:
##########
@@ -781,43 +889,6 @@ private void
setupStreamsAssignmentInterfaceWithTwoSubtopologies(final String su
);
}
- private void
setupStreamsAssignmentInterfaceWithTwoConcatenedSubtopologies(final String
subtopologyId1,
-
final String topicName1,
-
final String subtopologyId2,
-
final String topicName2) {
- when(streamsAssignmentInterface.subtopologyMap()).thenReturn(
- mkMap(
- mkEntry(
- subtopologyId1,
- new StreamsAssignmentInterface.Subtopology(
- Set.of(topicName1),
- Collections.emptySet(),
- Collections.emptyMap(),
- Collections.emptyMap(),
- Collections.emptyList()
- )
- ),
- mkEntry(
- subtopologyId2,
- new StreamsAssignmentInterface.Subtopology(
- Set.of(topicName2),
- Collections.emptySet(),
- mkMap(mkEntry(
- topicName2,
- new StreamsAssignmentInterface.TopicInfo(
- Optional.empty(),
- Optional.empty(),
- Collections.emptyMap()
- )
- )),
- Collections.emptyMap(),
- Collections.emptyList()
- )
- )
- )
- );
- }
-
Review Comment:
We removed an assignment with 2 subtopologies - will we add it back at some
point or is present in another form?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -422,6 +439,33 @@ private void process(final
ShareAcknowledgementCommitCallbackRegistrationEvent e
manager.setAcknowledgementCommitCallbackRegistered(event.isCallbackRegistered());
}
+ private void process(final StreamsOnTasksRevokedCallbackCompletedEvent
event) {
+ if (!requestManagers.streamsMembershipManager.isPresent()) {
Review Comment:
Maybe use `requestManagers.streamsMembershipManager.isEmpty()` instead here
and below.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java:
##########
@@ -281,17 +300,173 @@ public String toString() {
}
}
+ private final BlockingQueue<BackgroundEvent> onCallbackRequests = new
LinkedBlockingQueue<>();
+
+ private ApplicationEventHandler applicationEventHandler = null;
+
+ private Optional<Function<Set<StreamsAssignmentInterface.TaskId>,
Optional<Exception>>> onTasksRevokedCallback = null;
+ private Optional<Function<Assignment, Optional<Exception>>>
onTasksAssignedCallback = null;
+ private Optional<Supplier<Optional<Exception>>> onAllTasksLostCallback =
null;
+
+ private final StreamsRebalanceEventProcessor
streamsRebalanceEventProcessor;
+
+ private class StreamsRebalanceEventProcessor implements
EventProcessor<BackgroundEvent> {
+
+ @Override
+ public void process(final BackgroundEvent event) {
+ switch (event.type()) {
+ case ERROR:
+ process((ErrorEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED:
+ process((StreamsOnTasksRevokedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED:
+ process((StreamsOnTasksAssignedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED:
+ process((StreamsOnAllTasksLostCallbackNeededEvent) event);
+ break;
+
+ default:
+ throw new IllegalArgumentException("Background event type
" + event.type() + " was not expected");
+
+ }
+ }
+
+ private void process(final ErrorEvent event) {
+ throw event.error();
+ }
+
+ private void process(final StreamsOnTasksRevokedCallbackNeededEvent
event) {
Review Comment:
suggestion: refactor the method name to `processOnTasksRevokedCallbackNeeded`
Same as below for the other two methods - but this is a personal preference,
so feel free to leave as is.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java:
##########
@@ -281,17 +300,173 @@ public String toString() {
}
}
+ private final BlockingQueue<BackgroundEvent> onCallbackRequests = new
LinkedBlockingQueue<>();
+
+ private ApplicationEventHandler applicationEventHandler = null;
+
+ private Optional<Function<Set<StreamsAssignmentInterface.TaskId>,
Optional<Exception>>> onTasksRevokedCallback = null;
+ private Optional<Function<Assignment, Optional<Exception>>>
onTasksAssignedCallback = null;
+ private Optional<Supplier<Optional<Exception>>> onAllTasksLostCallback =
null;
+
+ private final StreamsRebalanceEventProcessor
streamsRebalanceEventProcessor;
+
+ private class StreamsRebalanceEventProcessor implements
EventProcessor<BackgroundEvent> {
+
+ @Override
+ public void process(final BackgroundEvent event) {
+ switch (event.type()) {
+ case ERROR:
+ process((ErrorEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED:
+ process((StreamsOnTasksRevokedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED:
+ process((StreamsOnTasksAssignedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED:
+ process((StreamsOnAllTasksLostCallbackNeededEvent) event);
+ break;
+
+ default:
+ throw new IllegalArgumentException("Background event type
" + event.type() + " was not expected");
+
+ }
+ }
+
+ private void process(final ErrorEvent event) {
+ throw event.error();
+ }
+
+ private void process(final StreamsOnTasksRevokedCallbackNeededEvent
event) {
+ StreamsOnTasksRevokedCallbackCompletedEvent invokedEvent =
invokeOnTasksRevokedCallback(event.activeTasksToRevoke(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnTasksAssignedCallbackNeededEvent
event) {
+ StreamsOnTasksAssignedCallbackCompletedEvent invokedEvent =
invokeOnTasksAssignedCallback(event.assignment(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnAllTasksLostCallbackNeededEvent
event) {
+ StreamsOnAllTasksLostCallbackCompletedEvent invokedEvent =
invokeOnAllTasksLostCallback(event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private StreamsOnTasksRevokedCallbackCompletedEvent
invokeOnTasksRevokedCallback(final Set<StreamsAssignmentInterface.TaskId>
activeTasksToRevoke,
+
final CompletableFuture<Void> future) {
+ final Optional<Exception> exceptionFromCallback =
onTasksRevokedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).apply(activeTasksToRevoke);
+
+ return exceptionFromCallback
+ .map(exception ->
+ new StreamsOnTasksRevokedCallbackCompletedEvent(
+ future,
+
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exception, "Task revocation
callback throws an error"))
+ ))
+ .orElseGet(() -> new
StreamsOnTasksRevokedCallbackCompletedEvent(future, Optional.empty()));
+ }
+
+ private StreamsOnTasksAssignedCallbackCompletedEvent
invokeOnTasksAssignedCallback(final StreamsAssignmentInterface.Assignment
assignment,
+
final CompletableFuture<Void> future) {
+ Optional<KafkaException> error = Optional.empty();
+ // ToDo: Can we avoid the following check?
+ if (!assignment.equals(reconciledAssignment.get())) {
+
+ final Optional<Exception> exceptionFromCallback =
onTasksAssignedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).apply(assignment);
+
+ if (exceptionFromCallback.isPresent()) {
+ error =
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(),
"Task assignment callback throws an error"));
+ } else {
+ reconciledAssignment.set(assignment);
+ }
Review Comment:
Maybe use `Optional.ifPresentOrElse` ?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -709,6 +708,11 @@ public StreamThread(final Time time,
this.streamsUncaughtExceptionHandler = streamsUncaughtExceptionHandler;
this.cacheResizer = cacheResizer;
this.streamsAssignmentInterface = streamsAssignmentInterface;
+ if (streamsAssignmentInterface != null) {
Review Comment:
Maybe use an `Optional` for `streamsAssignmentInterface`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java:
##########
@@ -281,17 +300,173 @@ public String toString() {
}
}
+ private final BlockingQueue<BackgroundEvent> onCallbackRequests = new
LinkedBlockingQueue<>();
+
+ private ApplicationEventHandler applicationEventHandler = null;
+
+ private Optional<Function<Set<StreamsAssignmentInterface.TaskId>,
Optional<Exception>>> onTasksRevokedCallback = null;
+ private Optional<Function<Assignment, Optional<Exception>>>
onTasksAssignedCallback = null;
+ private Optional<Supplier<Optional<Exception>>> onAllTasksLostCallback =
null;
+
+ private final StreamsRebalanceEventProcessor
streamsRebalanceEventProcessor;
+
+ private class StreamsRebalanceEventProcessor implements
EventProcessor<BackgroundEvent> {
+
+ @Override
+ public void process(final BackgroundEvent event) {
+ switch (event.type()) {
+ case ERROR:
+ process((ErrorEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED:
+ process((StreamsOnTasksRevokedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED:
+ process((StreamsOnTasksAssignedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED:
+ process((StreamsOnAllTasksLostCallbackNeededEvent) event);
+ break;
+
+ default:
+ throw new IllegalArgumentException("Background event type
" + event.type() + " was not expected");
+
+ }
+ }
+
+ private void process(final ErrorEvent event) {
+ throw event.error();
+ }
+
+ private void process(final StreamsOnTasksRevokedCallbackNeededEvent
event) {
+ StreamsOnTasksRevokedCallbackCompletedEvent invokedEvent =
invokeOnTasksRevokedCallback(event.activeTasksToRevoke(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnTasksAssignedCallbackNeededEvent
event) {
+ StreamsOnTasksAssignedCallbackCompletedEvent invokedEvent =
invokeOnTasksAssignedCallback(event.assignment(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnAllTasksLostCallbackNeededEvent
event) {
+ StreamsOnAllTasksLostCallbackCompletedEvent invokedEvent =
invokeOnAllTasksLostCallback(event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private StreamsOnTasksRevokedCallbackCompletedEvent
invokeOnTasksRevokedCallback(final Set<StreamsAssignmentInterface.TaskId>
activeTasksToRevoke,
+
final CompletableFuture<Void> future) {
+ final Optional<Exception> exceptionFromCallback =
onTasksRevokedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).apply(activeTasksToRevoke);
+
+ return exceptionFromCallback
+ .map(exception ->
+ new StreamsOnTasksRevokedCallbackCompletedEvent(
+ future,
+
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exception, "Task revocation
callback throws an error"))
+ ))
+ .orElseGet(() -> new
StreamsOnTasksRevokedCallbackCompletedEvent(future, Optional.empty()));
+ }
+
+ private StreamsOnTasksAssignedCallbackCompletedEvent
invokeOnTasksAssignedCallback(final StreamsAssignmentInterface.Assignment
assignment,
+
final CompletableFuture<Void> future) {
+ Optional<KafkaException> error = Optional.empty();
+ // ToDo: Can we avoid the following check?
+ if (!assignment.equals(reconciledAssignment.get())) {
+
+ final Optional<Exception> exceptionFromCallback =
onTasksAssignedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).apply(assignment);
+
+ if (exceptionFromCallback.isPresent()) {
+ error =
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(),
"Task assignment callback throws an error"));
+ } else {
+ reconciledAssignment.set(assignment);
+ }
+ }
+ return new StreamsOnTasksAssignedCallbackCompletedEvent(future,
error);
+ }
+
+ private StreamsOnAllTasksLostCallbackCompletedEvent
invokeOnAllTasksLostCallback(final CompletableFuture<Void> future) {
+ final Optional<Exception> exceptionFromCallback =
onAllTasksLostCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).get();
+
+ final Optional<KafkaException> error;
+
+ if (exceptionFromCallback.isPresent()) {
+ error =
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(),
"Task assignment callback throws an error"));
+ } else {
+ error = Optional.empty();
+ reconciledAssignment.set(Assignment.EMPTY);
+ }
Review Comment:
Same here
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java:
##########
@@ -281,17 +300,173 @@ public String toString() {
}
}
+ private final BlockingQueue<BackgroundEvent> onCallbackRequests = new
LinkedBlockingQueue<>();
+
+ private ApplicationEventHandler applicationEventHandler = null;
+
+ private Optional<Function<Set<StreamsAssignmentInterface.TaskId>,
Optional<Exception>>> onTasksRevokedCallback = null;
+ private Optional<Function<Assignment, Optional<Exception>>>
onTasksAssignedCallback = null;
+ private Optional<Supplier<Optional<Exception>>> onAllTasksLostCallback =
null;
+
+ private final StreamsRebalanceEventProcessor
streamsRebalanceEventProcessor;
+
+ private class StreamsRebalanceEventProcessor implements
EventProcessor<BackgroundEvent> {
+
+ @Override
+ public void process(final BackgroundEvent event) {
+ switch (event.type()) {
+ case ERROR:
+ process((ErrorEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED:
+ process((StreamsOnTasksRevokedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED:
+ process((StreamsOnTasksAssignedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED:
+ process((StreamsOnAllTasksLostCallbackNeededEvent) event);
+ break;
+
+ default:
+ throw new IllegalArgumentException("Background event type
" + event.type() + " was not expected");
+
+ }
+ }
+
+ private void process(final ErrorEvent event) {
+ throw event.error();
+ }
+
+ private void process(final StreamsOnTasksRevokedCallbackNeededEvent
event) {
+ StreamsOnTasksRevokedCallbackCompletedEvent invokedEvent =
invokeOnTasksRevokedCallback(event.activeTasksToRevoke(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnTasksAssignedCallbackNeededEvent
event) {
+ StreamsOnTasksAssignedCallbackCompletedEvent invokedEvent =
invokeOnTasksAssignedCallback(event.assignment(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnAllTasksLostCallbackNeededEvent
event) {
+ StreamsOnAllTasksLostCallbackCompletedEvent invokedEvent =
invokeOnAllTasksLostCallback(event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private StreamsOnTasksRevokedCallbackCompletedEvent
invokeOnTasksRevokedCallback(final Set<StreamsAssignmentInterface.TaskId>
activeTasksToRevoke,
+
final CompletableFuture<Void> future) {
+ final Optional<Exception> exceptionFromCallback =
onTasksRevokedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).apply(activeTasksToRevoke);
+
+ return exceptionFromCallback
+ .map(exception ->
+ new StreamsOnTasksRevokedCallbackCompletedEvent(
+ future,
+
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exception, "Task revocation
callback throws an error"))
+ ))
+ .orElseGet(() -> new
StreamsOnTasksRevokedCallbackCompletedEvent(future, Optional.empty()));
+ }
+
+ private StreamsOnTasksAssignedCallbackCompletedEvent
invokeOnTasksAssignedCallback(final StreamsAssignmentInterface.Assignment
assignment,
+
final CompletableFuture<Void> future) {
+ Optional<KafkaException> error = Optional.empty();
+ // ToDo: Can we avoid the following check?
+ if (!assignment.equals(reconciledAssignment.get())) {
+
+ final Optional<Exception> exceptionFromCallback =
onTasksAssignedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).apply(assignment);
+
+ if (exceptionFromCallback.isPresent()) {
+ error =
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(),
"Task assignment callback throws an error"));
+ } else {
+ reconciledAssignment.set(assignment);
+ }
+ }
+ return new StreamsOnTasksAssignedCallbackCompletedEvent(future,
error);
+ }
+
+ private StreamsOnAllTasksLostCallbackCompletedEvent
invokeOnAllTasksLostCallback(final CompletableFuture<Void> future) {
+ final Optional<Exception> exceptionFromCallback =
onAllTasksLostCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks
assignment callback set!")).get();
+
+ final Optional<KafkaException> error;
+
+ if (exceptionFromCallback.isPresent()) {
+ error =
Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(),
"Task assignment callback throws an error"));
+ } else {
+ error = Optional.empty();
+ reconciledAssignment.set(Assignment.EMPTY);
+ }
+
+ return new StreamsOnAllTasksLostCallbackCompletedEvent(future,
error);
+ }
+ }
+
public StreamsAssignmentInterface(UUID processId,
Optional<HostInfo> endpoint,
Map<String, Subtopology> subtopologyMap,
- Map<String, String> clientTags
- ) {
+ Map<String, String> clientTags) {
this.processId = processId;
this.endpoint = endpoint;
this.subtopologyMap = subtopologyMap;
this.taskLags = new HashMap<>();
this.shutdownRequested = new AtomicBoolean(false);
this.clientTags = clientTags;
+ this.streamsRebalanceEventProcessor = new
StreamsRebalanceEventProcessor();
+ }
+
+ public void setOnTasksRevokedCallback(final
Function<Set<StreamsAssignmentInterface.TaskId>, Optional<Exception>>
onTasksRevokedCallback) {
+ this.onTasksRevokedCallback =
Optional.ofNullable(onTasksRevokedCallback);
+ }
+
+ public void setOnTasksAssignedCallback(final Function<Assignment,
Optional<Exception>> onTasksAssignedCallback) {
+ this.onTasksAssignedCallback =
Optional.ofNullable(onTasksAssignedCallback);
+ }
+
+ public void setOnAllTasksLostCallback(final Supplier<Optional<Exception>>
onAllTasksLostCallback) {
+ this.onAllTasksLostCallback =
Optional.ofNullable(onAllTasksLostCallback);
+ }
+
+ public void setApplicationEventHandler(final ApplicationEventHandler
applicationEventHandler) {
+ this.applicationEventHandler = applicationEventHandler;
+ }
+
+ public CompletableFuture<Void>
requestOnTasksAssignedCallbackInvocation(final Assignment assignment) {
+ final StreamsOnTasksAssignedCallbackNeededEvent
onTasksAssignedCallbackNeededEvent = new
StreamsOnTasksAssignedCallbackNeededEvent(assignment);
+ onCallbackRequests.add(onTasksAssignedCallbackNeededEvent);
+ return onTasksAssignedCallbackNeededEvent.future();
+ }
+
+ public CompletableFuture<Void>
requestOnTasksRevokedCallbackInvocation(final
Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke) {
+ final StreamsOnTasksRevokedCallbackNeededEvent
onTasksRevokedCallbackNeededEvent = new
StreamsOnTasksRevokedCallbackNeededEvent(activeTasksToRevoke);
+ onCallbackRequests.add(onTasksRevokedCallbackNeededEvent);
+ return onTasksRevokedCallbackNeededEvent.future();
+ }
+
+ public CompletableFuture<Void> requestOnAllTasksLostCallbackInvocation() {
Review Comment:
Why all three `CompletableFuture` have no return value? I guess we're saying
that `CompletableFuture.get()` doesn't throw an `ExcecutionException`
everything succeeded?
Asking for my own understanding
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
