KAFKA-9113, P2: Extract Producer to RecordCollector

clients/src/main/java/org/apache/kafka/clients/producer/Callback.java

@@ -39,6 +39,7 @@ public interface Callback {
      *                  RecordBatchTooLargeException
      *                  RecordTooLargeException
      *                  UnknownServerException
+     *                  UnknownProducerIdException
      *
      *                  Retriable exceptions (transient, may be covered by increasing #.retries):
      *

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -1149,7 +1149,9 @@ public List<PartitionInfo> partitionsFor(String topic) {
      * block forever.</strong>
      * <p>
      *
-     * @throws InterruptException If the thread is interrupted while blocked
+     * @throws InterruptException If the thread is interrupted while blocked.
+     * @throws KafkaException If a unexpected error occurs while trying to close the client, this error should be treated
+     *                        as fatal and indicate the client is no longer functionable.
      */
     @Override
     public void close() {
@@ -1169,7 +1171,9 @@ public void close() {
      *
      * @param timeout The maximum time to wait for producer to complete any pending requests. The value should be
      *                non-negative. Specifying a timeout of zero means do not wait for pending send requests to complete.
-     * @throws InterruptException If the thread is interrupted while blocked
+     * @throws InterruptException If the thread is interrupted while blocked.
+     * @throws KafkaException If a unexpected error occurs while trying to close the client, this error should be treated
+     *                        as fatal and indicate the client is no longer functionable.
      * @throws IllegalArgumentException If the <code>timeout</code> is negative.
      *
      */

streams/src/main/java/org/apache/kafka/streams/errors/TaskMigratedException.java

@@ -18,7 +18,7 @@
 
 
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.streams.processor.internals.Task;
+import org.apache.kafka.streams.processor.TaskId;
 
 /**
  * Indicates that a task got migrated to another thread.
@@ -28,43 +28,41 @@ public class TaskMigratedException extends StreamsException {
 
     private final static long serialVersionUID = 1L;
 
-    private final Task task;
+    private final TaskId taskId;
 
-    // this is for unit test only
-    public TaskMigratedException() {
-        super("A task has been migrated unexpectedly", null);
-
-        this.task = null;
-    }
-
-    public TaskMigratedException(final Task task,
+    public TaskMigratedException(final TaskId taskId,
                                  final TopicPartition topicPartition,
                                  final long endOffset,
                                  final long pos) {
-        super(String.format("Log end offset of %s should not change while restoring: old end offset %d, current offset %d",
-                            topicPartition,
-                            endOffset,
-                            pos),
-            null);
-
-        this.task = task;
+        this(taskId, String.format("Log end offset of %s should not change while restoring: old end offset %d, current offset %d",
+            topicPartition,
+            endOffset,
+            pos), null);
     }
 
-    public TaskMigratedException(final Task task) {
-        super(String.format("Task %s is unexpectedly closed during processing", task.id()), null);
-
-        this.task = task;
+    public TaskMigratedException(final TaskId taskId) {
+        this(taskId, String.format("Task %s is unexpectedly closed during processing", taskId), null);
     }
 
-    public TaskMigratedException(final Task task,
+    public TaskMigratedException(final TaskId taskId,
                                  final Throwable throwable) {
-        super(String.format("Client request for task %s has been fenced due to a rebalance", task.id()), throwable);
+        this(taskId, String.format("Client request for task %s has been fenced due to a rebalance", taskId), throwable);
+    }
 
-        this.task = task;
+    public TaskMigratedException(final TaskId taskId,
+                                 final String message,
+                                 final Throwable throwable) {
+        super(message, throwable);
+        this.taskId = taskId;
     }
 
-    public Task migratedTask() {
-        return task;
+    public TaskId migratedTaskId() {
+        return taskId;
     }
 
+    // this is for unit test only
+    // TODO K9113: remove this after we've refactored AssignedTasksTests
+    public TaskMigratedException() {
+        this(null, "A task has been migrated unexpectedly", null);
+    }
 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

@@ -23,7 +23,6 @@
 import org.apache.kafka.streams.errors.LockException;
 import org.apache.kafka.streams.errors.ProcessorStateException;
 import org.apache.kafka.streams.errors.StreamsException;
-import org.apache.kafka.streams.errors.TaskMigratedException;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.TaskId;
@@ -43,7 +42,6 @@ public abstract class AbstractTask implements Task {
     final Set<TopicPartition> partitions;
     final Consumer<byte[], byte[]> consumer;
     final String logPrefix;
-    final boolean eosEnabled;
     final Logger log;
     final LogContext logContext;
     final StateDirectory stateDirectory;
@@ -61,37 +59,23 @@ public abstract class AbstractTask implements Task {
                  final Set<TopicPartition> partitions,
                  final ProcessorTopology topology,
                  final Consumer<byte[], byte[]> consumer,
-                 final ChangelogReader changelogReader,
                  final boolean isStandby,
+                 final ProcessorStateManager stateMgr,
                  final StateDirectory stateDirectory,
                  final StreamsConfig config) {
         this.id = id;
         this.applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);
         this.partitions = new HashSet<>(partitions);
         this.topology = topology;
         this.consumer = consumer;
-        this.eosEnabled = StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG));
         this.stateDirectory = stateDirectory;
 
+        this.stateMgr = stateMgr;
+
         final String threadIdPrefix = String.format("stream-thread [%s] ", Thread.currentThread().getName());
         this.logPrefix = threadIdPrefix + String.format("%s [%s] ", isStandby ? "standby-task" : "task", id);
         this.logContext = new LogContext(logPrefix);
         this.log = logContext.logger(getClass());
-
-        // create the processor state manager
-        try {
-            stateMgr = new ProcessorStateManager(
-                id,
-                partitions,
-                isStandby,
-                stateDirectory,
-                topology.storeToChangelogTopic(),
-                changelogReader,
-                eosEnabled,
-                logContext);
-        } catch (final IOException e) {
-            throw new ProcessorStateException(String.format("%sError while creating the state manager", logPrefix), e);
-        }
     }
 
     @Override
@@ -135,10 +119,6 @@ public String toString() {
         return toString("");
     }
 
-    public boolean isEosEnabled() {
-        return eosEnabled;
-    }
-
     /**
      * Produces a string representation containing useful information about a Task starting with the given indent.
      * This is useful in debugging scenarios.
@@ -169,19 +149,6 @@ public String toString(final String indent) {
         return sb.toString();
     }
 
-    /**
-     * Flush all state stores owned by this task
-     */
-    void flushState() {
-        try {
-            stateMgr.flush();
-        } catch (final ProcessorStateException e) {
-            if (e.getCause() instanceof RecoverableClientException) {
-                throw new TaskMigratedException(this, e);
-            }
-        }
-    }
-
     /**
      * Package-private for testing only
      *

streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStandbyTasks.java

@@ -76,7 +76,7 @@ List<TopicPartition> closeRevokedStandbyTasks(final Map<TaskId, Set<TopicPartiti
             }
 
             try {
-                task.close(true, false);
+                task.close(true);
             } catch (final RuntimeException e) {
                 log.error("Closing the standby task {} failed due to the following error:", task.id(), e);
             } finally {

streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java

@@ -75,9 +75,9 @@ boolean allTasksRunning() {
     @Override
     void closeTask(final StreamTask task, final boolean clean) {
         if (suspended.containsKey(task.id())) {
-            task.closeSuspended(clean, null);
+            task.closeSuspended(clean);
         } else {
-            task.close(clean, false);
+            task.close(clean);
         }
     }
 
@@ -155,7 +155,7 @@ private RuntimeException suspendRunningTasks(final Set<TaskId> runningTasksToSus
                 firstException.compareAndSet(null, e);
                 try {
                     prevActiveTasks.remove(id);
-                    task.close(false, false);
+                    task.close(false);
                 } catch (final RuntimeException f) {
                     log.error(
                         "After suspending failed, closing the same stream task {} failed again due to the following error:",
@@ -179,7 +179,7 @@ private RuntimeException closeNonRunningTasks(final Set<TaskId> nonRunningTasksT
 
         for (final TaskId id : nonRunningTasksToClose) {
             final StreamTask task = created.get(id);
-            firstException.compareAndSet(null, closeNonRunning(false, task, closedTaskChangelogs));
+            firstException.compareAndSet(null, closeNonRunning(true, task, closedTaskChangelogs));
         }
 
         return firstException.get();
@@ -192,19 +192,17 @@ RuntimeException closeRestoringTasks(final Set<TaskId> restoringTasksToClose,
 
         for (final TaskId id : restoringTasksToClose) {
             final StreamTask task = restoring.get(id);
-            firstException.compareAndSet(null, closeRestoring(false, task, closedTaskChangelogs));
+            firstException.compareAndSet(null, closeRestoring(true, task, closedTaskChangelogs));
         }
 
         return firstException.get();
     }
 
-    private RuntimeException closeRunning(final boolean isZombie,
-                                          final StreamTask task) {
+    private RuntimeException closeRunning(final boolean clean, final StreamTask task) {
         removeTaskFromAllStateMaps(task, Collections.emptyMap());
 
         try {
-            final boolean clean = !isZombie;
-            task.close(clean, isZombie);
+            task.close(clean);
         } catch (final RuntimeException e) {
             log.error("Failed to close the stream task {}", task.id(), e);
             return e;
@@ -213,14 +211,14 @@ private RuntimeException closeRunning(final boolean isZombie,
         return null;
     }
 
-    private RuntimeException closeNonRunning(final boolean isZombie,
+    private RuntimeException closeNonRunning(final boolean clean,
                                              final StreamTask task,
                                              final List<TopicPartition> closedTaskChangelogs) {
         removeTaskFromAllStateMaps(task, Collections.emptyMap());
         closedTaskChangelogs.addAll(task.changelogPartitions());
 
         try {
-            task.close(false, isZombie);
+            task.close(clean);
         } catch (final RuntimeException e) {
             log.error("Failed to close the stream task {}", task.id(), e);
             return e;
@@ -230,14 +228,13 @@ private RuntimeException closeNonRunning(final boolean isZombie,
     }
 
     // Since a restoring task has not had its topology initialized yet, we need only close the state manager
-    private RuntimeException closeRestoring(final boolean isZombie,
+    private RuntimeException closeRestoring(final boolean clean,
                                             final StreamTask task,
                                             final List<TopicPartition> closedTaskChangelogs) {
         removeTaskFromAllStateMaps(task, Collections.emptyMap());
         closedTaskChangelogs.addAll(task.changelogPartitions());
 
         try {
-            final boolean clean = !isZombie;
             task.closeStateManager(clean);
         } catch (final RuntimeException e) {
             log.error("Failed to close the restoring stream task {} due to the following error:", task.id(), e);
@@ -247,13 +244,11 @@ private RuntimeException closeRestoring(final boolean isZombie,
         return null;
     }
 
-    private RuntimeException closeSuspended(final boolean isZombie,
-                                            final StreamTask task) {
+    private RuntimeException closeSuspended(final boolean clean, final StreamTask task) {
         removeTaskFromAllStateMaps(task, Collections.emptyMap());
 
         try {
-            final boolean clean = !isZombie;
-            task.closeSuspended(clean, null);
+            task.closeSuspended(clean);
         } catch (final RuntimeException e) {
             log.error("Failed to close the suspended stream task {} due to the following error:", task.id(), e);
             return e;
@@ -270,7 +265,7 @@ RuntimeException closeNotAssignedSuspendedTasks(final Set<TaskId> revokedTasks)
             final StreamTask suspendedTask = suspended.get(revokedTask);
 
             if (suspendedTask != null) {
-                firstException.compareAndSet(null, closeSuspended(false, suspendedTask));
+                firstException.compareAndSet(null, closeSuspended(true, suspendedTask));
             } else {
                 log.debug("Revoked stream task {} could not be found in suspended, may have already been closed", revokedTask);
             }
@@ -287,16 +282,16 @@ RuntimeException closeAllTasksAsZombies() {
         for (final TaskId id : allAssignedTaskIds()) {
             if (running.containsKey(id)) {
                 log.debug("Closing the zombie running stream task {}.", id);
-                firstException.compareAndSet(null, closeRunning(true, running.get(id)));
+                firstException.compareAndSet(null, closeRunning(false, running.get(id)));
             } else if (created.containsKey(id)) {
                 log.debug("Closing the zombie created stream task {}.", id);
-                firstException.compareAndSet(null, closeNonRunning(true, created.get(id), changelogs));
+                firstException.compareAndSet(null, closeNonRunning(false, created.get(id), changelogs));
             } else if (restoring.containsKey(id)) {
                 log.debug("Closing the zombie restoring stream task {}.", id);
-                firstException.compareAndSet(null, closeRestoring(true, restoring.get(id), changelogs));
+                firstException.compareAndSet(null, closeRestoring(false, restoring.get(id), changelogs));
             } else if (suspended.containsKey(id)) {
                 log.debug("Closing the zombie suspended stream task {}.", id);
-                firstException.compareAndSet(null, closeSuspended(true, suspended.get(id)));
+                firstException.compareAndSet(null, closeSuspended(false, suspended.get(id)));
             }
         }
 
@@ -330,7 +325,7 @@ boolean maybeResumeSuspendedTask(final TaskId taskId,
                 return true;
             } else {
                 log.warn("Couldn't resume stream task {} assigned partitions {}, task partitions {}", taskId, partitions, task.partitions());
-                task.closeSuspended(true, null);
+                task.closeSuspended(true);
             }
         }
         return false;

streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java

@@ -99,7 +99,7 @@ Collection<T> running() {
 
     void tryCloseZombieTask(final T task) {
         try {
-            task.close(false, true);
+            task.close(false);
         } catch (final RuntimeException e) {
             log.warn("Failed to close zombie {} {} due to {}; ignore and proceed.", taskTypeName, task.id(), e.toString());
         }
@@ -279,13 +279,13 @@ void shutdown(final boolean clean) {
     }
 
     void closeTask(final T task, final boolean clean) {
-        task.close(clean, false);
+        task.close(clean);
     }
 
     private void closeUnclean(final T task) {
         log.info("Try to close {} {} unclean.", task.getClass().getSimpleName(), task.id());
         try {
-            task.close(false, false);
+            task.close(false);
         } catch (final RuntimeException fatalException) {
             log.error("Failed while closing {} {} due to the following error:",
                 task.getClass().getSimpleName(),

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java

@@ -163,7 +163,6 @@ public void reinitializeStateStoresForPartitions(final Collection<TopicPartition
                                                      final InternalProcessorContext processorContext) {
         StateManagerUtil.reinitializeStateStoresForPartitions(
             log,
-            eosEnabled,
             baseDir,
             globalStores,
             topology.storeToChangelogTopic(),