Add javadocs

Author: cadonna

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupRebalanceCallbacks.java

@@ -19,11 +19,31 @@
 import java.util.Optional;
 import java.util.Set;
 
+/**
+ * Callbacks for handling Streams group rebalance events in Kafka Streams.
+ */
 public interface StreamsGroupRebalanceCallbacks {
 
+    /**
+     * Called when tasks are revoked from a stream thread.
+     *
+     * @param tasks The tasks to be revoked.
+     * @return The exception thrown during the callback, if any.
+     */
     Optional<Exception> onTasksRevoked(final Set<StreamsRebalanceData.TaskId> tasks);
 
+    /**
+     * Called when tasks are assigned from a stream thread.
+     *
+     * @param assignment The tasks assigned.
+     * @return The exception thrown during the callback, if any.
+     */
     Optional<Exception> onTasksAssigned(final StreamsRebalanceData.Assignment assignment);
 
+    /**
+     * Called when a stream thread loses all assigned tasks.
+     *
+     * @return The exception thrown during the callback, if any.
+     */
     Optional<Exception> onAllTasksLost();
 }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

@@ -24,6 +24,9 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
+/**
+ * This class holds the data that is needed to participate in the Streams rebalance protocol.
+ */
 public class StreamsRebalanceData {
 
     public static class Assignment {
@@ -187,13 +190,7 @@ public String toString() {
 
     private final Map<String, Subtopology> subtopologies;
 
-    private final AtomicReference<Assignment> reconciledAssignment = new AtomicReference<>(
-        new Assignment(
-            new HashSet<>(),
-            new HashSet<>(),
-            new HashSet<>()
-        )
-    );
+    private final AtomicReference<Assignment> reconciledAssignment = new AtomicReference<>(Assignment.EMPTY);
 
     public StreamsRebalanceData(Map<String, Subtopology> subtopologies) {
         this.subtopologies = subtopologies;

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessor.java

@@ -34,37 +34,75 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.LinkedBlockingQueue;
 
+/**
+ * Processes events from the Streams rebalance protocol.
+ * <p>
+ * The Streams rebalance processor receives events from the background thread of the async consumer, more precisely
+ * from the Streams membership manager and handles them.
+ * For example, events are requests for invoking the task assignment and task revocation callbacks.
+ * Results of the event handling are passed back to the background thread.
+ */
 public class StreamsRebalanceEventsProcessor {
 
     private final BlockingQueue<BackgroundEvent> onCallbackRequests = new LinkedBlockingQueue<>();
     private ApplicationEventHandler applicationEventHandler = null;
     private final StreamsGroupRebalanceCallbacks rebalanceCallbacks;
     private final StreamsRebalanceData streamsRebalanceData;
 
+    /**
+     * Constructs the Streams rebalance processor.
+     *
+     * @param streamsRebalanceData
+     * @param rebalanceCallbacks
+     */
     public StreamsRebalanceEventsProcessor(StreamsRebalanceData streamsRebalanceData,
                                            StreamsGroupRebalanceCallbacks rebalanceCallbacks) {
         this.streamsRebalanceData = streamsRebalanceData;
         this.rebalanceCallbacks = rebalanceCallbacks;
     }
 
+    /**
+     * Requests the invocation of the task assignment callback.
+     *
+     * @param assignment The tasks to be assigned to the member of the Streams group.
+     * @return A future that will be completed when the callback has been invoked.
+     */
     public CompletableFuture<Void> requestOnTasksAssignedCallbackInvocation(final StreamsRebalanceData.Assignment assignment) {
         final StreamsOnTasksAssignedCallbackNeededEvent onTasksAssignedCallbackNeededEvent = new StreamsOnTasksAssignedCallbackNeededEvent(assignment);
         onCallbackRequests.add(onTasksAssignedCallbackNeededEvent);
         return onTasksAssignedCallbackNeededEvent.future();
     }
 
+    /**
+     * Requests the invocation of the task revocation callback.
+     *
+     * @param activeTasksToRevoke The tasks to revoke from the member of the Streams group
+     * @return A future that will be completed when the callback has been invoked.
+     */
     public CompletableFuture<Void> requestOnTasksRevokedCallbackInvocation(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke) {
         final StreamsOnTasksRevokedCallbackNeededEvent onTasksRevokedCallbackNeededEvent = new StreamsOnTasksRevokedCallbackNeededEvent(activeTasksToRevoke);
         onCallbackRequests.add(onTasksRevokedCallbackNeededEvent);
         return onTasksRevokedCallbackNeededEvent.future();
     }
 
+    /**
+     * Requests the invocation of the all tasks lost callback.
+     *
+     * @return A future that will be completed when the callback has been invoked.
+     */
     public CompletableFuture<Void> requestOnAllTasksLostCallbackInvocation() {
         final StreamsOnAllTasksLostCallbackNeededEvent onAllTasksLostCallbackNeededEvent = new StreamsOnAllTasksLostCallbackNeededEvent();
         onCallbackRequests.add(onAllTasksLostCallbackNeededEvent);
         return onAllTasksLostCallbackNeededEvent.future();
     }
 
+    /**
+     * Sets the application event handler.
+     *
+     * The application handler sends the results of the callbacks to the background thread.
+     *
+     * @param applicationEventHandler The application handler.
+     */
     public void setApplicationEventHandler(final ApplicationEventHandler applicationEventHandler) {
         this.applicationEventHandler = applicationEventHandler;
     }
@@ -153,6 +191,9 @@ private StreamsOnAllTasksLostCallbackCompletedEvent invokeOnAllTasksLostCallback
         return new StreamsOnAllTasksLostCallbackCompletedEvent(future, error);
     }
 
+    /**
+     * Processes all events received from the background thread so far.
+     */
     public void process() {
         LinkedList<BackgroundEvent> events = new LinkedList<>();
         onCallbackRequests.drainTo(events);