philipnee commented on code in PR #14406:
URL: https://github.com/apache/kafka/pull/14406#discussion_r1339158103
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -17,146 +17,195 @@
package org.apache.kafka.clients.consumer.internals.events;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.clients.consumer.internals.CachedSupplier;
import org.apache.kafka.clients.consumer.internals.CommitRequestManager;
import org.apache.kafka.clients.consumer.internals.ConsumerMetadata;
+import org.apache.kafka.clients.consumer.internals.DefaultBackgroundThread;
import org.apache.kafka.clients.consumer.internals.RequestManagers;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.utils.LogContext;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
-public class ApplicationEventProcessor {
-
- private final BlockingQueue<BackgroundEvent> backgroundEventQueue;
+/**
+ * An {@link EventProcessor} that is created and executes in {@link
DefaultBackgroundThread the background thread}
+ * which processes {@link ApplicationEvent application events} generated by
the application thread.
+ */
+public class ApplicationEventProcessor extends
EventProcessor<ApplicationEvent> {
private final ConsumerMetadata metadata;
-
private final RequestManagers requestManagers;
- public ApplicationEventProcessor(final BlockingQueue<BackgroundEvent>
backgroundEventQueue,
+ public ApplicationEventProcessor(final LogContext logContext,
+ final BlockingQueue<ApplicationEvent>
applicationEventQueue,
final RequestManagers requestManagers,
final ConsumerMetadata metadata) {
- this.backgroundEventQueue = backgroundEventQueue;
+ super(logContext, applicationEventQueue);
this.requestManagers = requestManagers;
this.metadata = metadata;
}
- public boolean process(final ApplicationEvent event) {
- Objects.requireNonNull(event);
+ /**
+ * Process the events—if any—that were produced by the application thread.
It is possible that when processing
+ * an event generates an error. In such cases, the processor will
immediately throw an exception, and not
+ * process the remaining events.
+ */
+ @Override
+ public void process() {
+ process(error -> {
+ throw error;
+ });
+ }
+
+ @Override
+ public void process(ApplicationEvent event) {
switch (event.type()) {
- case NOOP:
- return process((NoopApplicationEvent) event);
case COMMIT:
- return process((CommitApplicationEvent) event);
+ process((CommitApplicationEvent) event);
+ return;
+
case POLL:
- return process((PollApplicationEvent) event);
+ process((PollApplicationEvent) event);
+ return;
+
case FETCH_COMMITTED_OFFSET:
- return process((OffsetFetchApplicationEvent) event);
+ process((OffsetFetchApplicationEvent) event);
+ return;
+
case METADATA_UPDATE:
- return process((NewTopicsMetadataUpdateRequestEvent) event);
+ process((NewTopicsMetadataUpdateRequestEvent) event);
+ return;
+
case ASSIGNMENT_CHANGE:
- return process((AssignmentChangeApplicationEvent) event);
+ process((AssignmentChangeApplicationEvent) event);
+ return;
+
case TOPIC_METADATA:
- return process((TopicMetadataApplicationEvent) event);
+ process((TopicMetadataApplicationEvent) event);
+ return;
+
case LIST_OFFSETS:
- return process((ListOffsetsApplicationEvent) event);
+ process((ListOffsetsApplicationEvent) event);
+ return;
+
+ case FETCH:
+ process((FetchEvent) event);
+ return;
+
case RESET_POSITIONS:
- return processResetPositionsEvent();
+ processResetPositionsEvent();
+ return;
+
case VALIDATE_POSITIONS:
- return processValidatePositionsEvent();
+ processValidatePositionsEvent();
+ return;
+
+ default:
+ throw new IllegalArgumentException("Application event type " +
event.type() + " was not expected");
}
- return false;
}
- /**
- * Processes {@link NoopApplicationEvent} and enqueue a
- * {@link NoopBackgroundEvent}. This is intentionally left here for
- * demonstration purpose.
- *
- * @param event a {@link NoopApplicationEvent}
- */
- private boolean process(final NoopApplicationEvent event) {
- return backgroundEventQueue.add(new
NoopBackgroundEvent(event.message()));
+ @Override
+ protected Class<ApplicationEvent> getEventClass() {
+ return ApplicationEvent.class;
}
- private boolean process(final PollApplicationEvent event) {
+ private void process(final PollApplicationEvent event) {
if (!requestManagers.commitRequestManager.isPresent()) {
- return true;
+ return;
}
CommitRequestManager manager =
requestManagers.commitRequestManager.get();
manager.updateAutoCommitTimer(event.pollTimeMs());
- return true;
}
- private boolean process(final CommitApplicationEvent event) {
+ private void process(final CommitApplicationEvent event) {
if (!requestManagers.commitRequestManager.isPresent()) {
// Leaving this error handling here, but it is a bit strange as
the commit API should enforce the group.id
- // upfront so we should never get to this block.
+ // upfront, so we should never get to this block.
Exception exception = new KafkaException("Unable to commit offset.
Most likely because the group.id wasn't set");
event.future().completeExceptionally(exception);
- return false;
+ return;
}
CommitRequestManager manager =
requestManagers.commitRequestManager.get();
event.chain(manager.addOffsetCommitRequest(event.offsets()));
- return true;
}
- private boolean process(final OffsetFetchApplicationEvent event) {
+ private void process(final OffsetFetchApplicationEvent event) {
if (!requestManagers.commitRequestManager.isPresent()) {
event.future().completeExceptionally(new KafkaException("Unable to
fetch committed " +
"offset because the CommittedRequestManager is not
available. Check if group.id was set correctly"));
- return false;
+ return;
}
CommitRequestManager manager =
requestManagers.commitRequestManager.get();
event.chain(manager.addOffsetFetchRequest(event.partitions()));
- return true;
}
- private boolean process(final NewTopicsMetadataUpdateRequestEvent event) {
+ private void process(final NewTopicsMetadataUpdateRequestEvent ignored) {
metadata.requestUpdateForNewTopics();
- return true;
}
- private boolean process(final AssignmentChangeApplicationEvent event) {
+ private void process(final AssignmentChangeApplicationEvent event) {
if (!requestManagers.commitRequestManager.isPresent()) {
- return false;
+ return;
Review Comment:
Thanks for removing the boolean.
--
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
