checkpoint;
+
if (state() == State.CREATED) {
// the task is created and not initialized, just re-write the checkpoint file
- executeAndMaybeSwallow(clean, () -> {
- stateMgr.checkpoint(Collections.emptyMap());
- }, "state manager checkpoint", log);
-
- transitionTo(State.CLOSING);
+ checkpoint = Collections.emptyMap();
} else if (state() == State.RUNNING) {
closeTopology(clean);
if (clean) {
- commitState();
- // whenever we have successfully committed state, it is safe to checkpoint
- // the state as well no matter if EOS is enabled or not
- stateMgr.checkpoint(checkpointableOffsets());
+ stateMgr.flush();
+ recordCollector.flush();
+ checkpoint = checkpointableOffsets();
} else {
+ checkpoint = null; // `null` indicates to not write a checkpoint
executeAndMaybeSwallow(false, stateMgr::flush, "state manager flush", log);
}
-
- transitionTo(State.CLOSING);
} else if (state() == State.RESTORING) {
- executeAndMaybeSwallow(clean, () -> {
- stateMgr.flush();
- stateMgr.checkpoint(Collections.emptyMap());
- }, "state manager flush and checkpoint", log);
-
- transitionTo(State.CLOSING);
+ executeAndMaybeSwallow(clean, stateMgr::flush, "state manager flush", log);
+ checkpoint = Collections.emptyMap();
} else if (state() == State.SUSPENDED) {
- // do not need to commit / checkpoint, since when suspending we've already committed the state
- transitionTo(State.CLOSING);
+ // if `SUSPENDED` do not need to checkpoint, since when suspending we've already committed the state
+ checkpoint = null; // `null` indicates to not write a checkpoint
+ } else {
+ throw new IllegalStateException("Illegal state " + state() + " while prepare closing active task " + id);
}
- if (state() == State.CLOSING) {
- // if EOS is enabled, we wipe out the whole state store for unclean close
- // since they are invalid to use anymore
- final boolean wipeStateStore = !clean && !eosDisabled;
+ return checkpoint;
+ }
- // first close state manager (which is idempotent) then close the record collector (which could throw),
- // if the latter throws and we re-close dirty which would close the state manager again.
- executeAndMaybeSwallow(clean, () -> {
- StateManagerUtil.closeStateManager(log, logPrefix, clean,
- wipeStateStore, stateMgr, stateDirectory, TaskType.ACTIVE);
- }, "state manager close", log);
+ /**
+ *
+ * the following order must be followed:
+ * 1. checkpoint the state manager -- even if we crash before this step, EOS is still guaranteed
+ * 2. then if we are closing on EOS and dirty, wipe out the state store directory
+ * 3. finally release the state manager lock
+ *
+ */
+ private void close(final boolean clean,
+ final Map checkpoint) {
+ if (clean && checkpoint != null) {
+ executeAndMaybeSwallow(clean, () -> stateMgr.checkpoint(checkpoint), "state manager checkpoint", log);
+ }
- executeAndMaybeSwallow(clean, recordCollector::close, "record collector close", log);
- } else {
- throw new IllegalStateException("Illegal state " + state() + " while closing active task " + id);
+ switch (state()) {
+ case CREATED:
+ case RUNNING:
+ case RESTORING:
+ case SUSPENDED:
+ // if EOS is enabled, we wipe out the whole state store for unclean close
+ // since they are invalid to use anymore
+ final boolean wipeStateStore = !clean && !eosDisabled;
+
+ // first close state manager (which is idempotent) then close the record collector (which could throw),
+ // if the latter throws and we re-close dirty which would close the state manager again.
+ executeAndMaybeSwallow(
+ clean,
+ () -> StateManagerUtil.closeStateManager(
+ log,
+ logPrefix,
+ clean,
+ wipeStateStore,
+ stateMgr,
+ stateDirectory,
+ TaskType.ACTIVE
+ ),
+ "state manager close",
+ log);
+
+ executeAndMaybeSwallow(clean, recordCollector::close, "record collector close", log);
+
+ break;
+
+ default:
+ throw new IllegalStateException("Illegal state " + state() + " while closing active task " + id);
}
partitionGroup.close();
@@ -472,7 +524,7 @@ private void close(final boolean clean) {
* source topic partitions, or if it is enforced to be processable
*/
public boolean isProcessable(final long wallClockTime) {
- if (state() == State.CLOSED || state() == State.CLOSING) {
+ if (state() == State.CLOSED) {
// a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
// in either case we can just log it and move on without notifying the thread since the consumer
// would soon be updated to not return any records for this task anymore.
@@ -509,8 +561,7 @@ public boolean isProcessable(final long wallClockTime) {
* @return true if this method processes a record, false if it does not process a record.
* @throws TaskMigratedException if the task producer got fenced (EOS only)
*/
- @SuppressWarnings("unchecked")
- @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
public boolean process(final long wallClockTime) {
if (!isProcessable(wallClockTime)) {
return false;
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index cb0179c65cfa2..89da266f1b03e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -57,6 +57,8 @@
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.kafka.streams.StreamsConfig.EXACTLY_ONCE;
+
public class StreamThread extends Thread {
private final Admin adminClient;
@@ -338,7 +340,8 @@ public static StreamThread create(final InternalTopologyBuilder builder,
standbyTaskCreator,
builder,
adminClient,
- stateDirectory
+ stateDirectory,
+ EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG))
);
log.info("Creating consumer client");
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java
index 0324bf261e04a..26ee9f0a97599 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java
@@ -59,24 +59,27 @@ public class StreamsProducer {
public StreamsProducer(final Producer producer,
final boolean eosEnabled,
- final LogContext logContext,
- final String applicationId) {
- log = logContext.logger(getClass());
- logPrefix = logContext.logPrefix().trim();
-
+ final String applicationId,
+ final LogContext logContext) {
this.producer = Objects.requireNonNull(producer, "producer cannot be null");
- this.applicationId = applicationId;
this.eosEnabled = eosEnabled;
+ this.applicationId = applicationId;
+ if (eosEnabled && applicationId == null) {
+ throw new IllegalArgumentException("applicationId cannot be null if EOS is enabled");
+ }
+
+ log = Objects.requireNonNull(logContext, "logContext cannot be null").logger(getClass());
+ logPrefix = logContext.logPrefix().trim();
}
private String formatException(final String message) {
- return message + " [" + logPrefix + ", " + (eosEnabled ? "eos" : "alo") + "]";
+ return message + " [" + logPrefix + "]";
}
/**
* @throws IllegalStateException if EOS is disabled
*/
- public void initTransaction() {
+ void initTransaction() {
if (!eosEnabled) {
throw new IllegalStateException(formatException("EOS is disabled"));
}
@@ -88,7 +91,7 @@ public void initTransaction() {
transactionInitialized = true;
} catch (final TimeoutException exception) {
log.warn(
- "Timeout exception caught when initializing transactions. " +
+ "Timeout exception caught trying to initialize transactions. " +
"The broker is either slow or in bad state (like not having enough replicas) in " +
"responding to the request, or the connection to broker was interrupted sending " +
"the request or receiving the response. " +
@@ -100,34 +103,34 @@ public void initTransaction() {
throw exception;
} catch (final KafkaException exception) {
throw new StreamsException(
- formatException("Error encountered while initializing transactions"),
+ formatException("Error encountered trying to initialize transactions"),
exception
);
}
}
}
- private void maybeBeginTransaction() throws ProducerFencedException {
+ void maybeBeginTransaction() throws ProducerFencedException {
if (eosEnabled && !transactionInFlight) {
try {
producer.beginTransaction();
transactionInFlight = true;
} catch (final ProducerFencedException error) {
throw new TaskMigratedException(
- formatException("Producer get fenced trying to begin a new transaction"),
+ formatException("Producer got fenced trying to begin a new transaction"),
error
);
} catch (final KafkaException error) {
throw new StreamsException(
- formatException("Producer encounter unexpected error trying to begin a new transaction"),
+ formatException("Error encountered trying to begin a new transaction"),
error
);
}
}
}
- public Future send(final ProducerRecord record,
- final Callback callback) {
+ Future send(final ProducerRecord record,
+ final Callback callback) {
maybeBeginTransaction();
try {
return producer.send(record, callback);
@@ -137,12 +140,12 @@ public Future send(final ProducerRecord record,
// in this case we should throw its wrapped inner cause so that it can be
// captured and re-wrapped as TaskMigrationException
throw new TaskMigratedException(
- formatException("Producer cannot send records anymore since it got fenced"),
+ formatException("Producer got fenced trying to send a record"),
uncaughtException.getCause()
);
} else {
throw new StreamsException(
- formatException(String.format("Error encountered sending record to topic %s", record.topic())),
+ formatException(String.format("Error encountered trying to send record to topic %s", record.topic())),
uncaughtException
);
}
@@ -158,7 +161,7 @@ private static boolean isRecoverable(final KafkaException uncaughtException) {
* @throws IllegalStateException if EOS is disabled
* @throws TaskMigratedException
*/
- public void commitTransaction(final Map offsets) throws ProducerFencedException {
+ void commitTransaction(final Map offsets) throws ProducerFencedException {
if (!eosEnabled) {
throw new IllegalStateException(formatException("EOS is disabled"));
}
@@ -169,15 +172,15 @@ public void commitTransaction(final Map offse
transactionInFlight = false;
} catch (final ProducerFencedException error) {
throw new TaskMigratedException(
- formatException("Producer get fenced trying to commit a transaction"),
+ formatException("Producer got fenced trying to commit a transaction"),
error
);
} catch (final TimeoutException error) {
// TODO KIP-447: we can consider treating it as non-fatal and retry on the thread level
- throw new StreamsException(formatException("Timed out while committing a transaction"), error);
+ throw new StreamsException(formatException("Timed out trying to commit a transaction"), error);
} catch (final KafkaException error) {
throw new StreamsException(
- formatException("Producer encounter unexpected error trying to commit a transaction"),
+ formatException("Error encountered trying to commit a transaction"),
error
);
}
@@ -186,7 +189,7 @@ public void commitTransaction(final Map offse
/**
* @throws IllegalStateException if EOS is disabled
*/
- public void abortTransaction() throws ProducerFencedException {
+ void abortTransaction() throws ProducerFencedException {
if (!eosEnabled) {
throw new IllegalStateException(formatException("EOS is disabled"));
}
@@ -204,7 +207,7 @@ public void abortTransaction() throws ProducerFencedException {
// can be ignored: transaction got already aborted by brokers/transactional-coordinator if this happens
} catch (final KafkaException error) {
throw new StreamsException(
- formatException("Producer encounter unexpected error trying to abort a transaction"),
+ formatException("Error encounter trying to abort a transaction"),
error
);
}
@@ -212,15 +215,18 @@ public void abortTransaction() throws ProducerFencedException {
}
}
- public List partitionsFor(final String topic) throws TimeoutException {
+ List partitionsFor(final String topic) throws TimeoutException {
return producer.partitionsFor(topic);
}
- public void flush() {
+ void flush() {
producer.flush();
}
- // for testing only
+ boolean eosEnabled() {
+ return eosEnabled;
+ }
+
Producer kafkaProducer() {
return producer;
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
index 2bdce69b65262..34fc600f57cd3 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.errors.LockException;
import org.apache.kafka.streams.errors.StreamsException;
@@ -73,8 +74,7 @@ enum State {
RESTORING(2, 3, 4), // 1
RUNNING(3, 4), // 2
SUSPENDED(1, 4), // 3
- CLOSING(4, 5), // 4, we allow CLOSING to transit to itself to make close idempotent
- CLOSED(0); // 5, we allow CLOSED to transit to CREATED to handle corrupted tasks
+ CLOSED(0); // 4, we allow CLOSED to transit to CREATED to handle corrupted tasks
private final Set validTransitions = new HashSet<>();
@@ -125,34 +125,48 @@ enum TaskType {
boolean commitNeeded();
/**
- * @throws TaskMigratedException all the task has been migrated
* @throws StreamsException fatal error, should close the thread
*/
- void commit();
+ void prepareCommit();
+
+ void postCommit();
/**
* @throws TaskMigratedException all the task has been migrated
* @throws StreamsException fatal error, should close the thread
*/
- void suspend();
+ void prepareSuspend();
+ void suspend();
/**
+ *
* @throws StreamsException fatal error, should close the thread
*/
void resume();
/**
- * Close a task that we still own. Commit all progress and close the task gracefully.
+ * Prepare to close a task that we still own and prepare it for committing
* Throws an exception if this couldn't be done.
+ * Must be idempotent.
*
- * @throws TaskMigratedException all the task has been migrated
* @throws StreamsException fatal error, should close the thread
*/
- void closeClean();
+ Map prepareCloseClean();
+
+ /**
+ * Must be idempotent.
+ */
+ void closeClean(final Map checkpoint);
/**
- * Close a task that we may not own. Discard any uncommitted progress and close the task.
+ * Prepare to close a task that we may not own. Discard any uncommitted progress and close the task.
* Never throws an exception, but just makes all attempts to release resources while closing.
+ * Must be idempotent.
+ */
+ void prepareCloseDirty();
+
+ /**
+ * Must be idempotent.
*/
void closeDirty();
@@ -182,6 +196,10 @@ default Map purgeableOffsets() {
return Collections.emptyMap();
}
+ default Map committableOffsetsAndMetadata() {
+ return Collections.emptyMap();
+ }
+
default boolean process(final long wallClockTime) {
return false;
}
@@ -197,4 +215,5 @@ default boolean maybePunctuateStreamTime() {
default boolean maybePunctuateSystemTime() {
return false;
}
+
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
index b804a4a7d6429..071c7cb3b6312 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
@@ -21,7 +21,9 @@
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.DeleteRecordsResult;
import org.apache.kafka.clients.admin.RecordsToDelete;
+import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
@@ -71,6 +73,7 @@ public class TaskManager {
private final InternalTopologyBuilder builder;
private final Admin adminClient;
private final StateDirectory stateDirectory;
+ private final boolean eosEnabled;
private final Map tasks = new TreeMap<>();
// materializing this relationship because the lookup is on the hot path
@@ -93,7 +96,8 @@ public class TaskManager {
final StandbyTaskCreator standbyTaskCreator,
final InternalTopologyBuilder builder,
final Admin adminClient,
- final StateDirectory stateDirectory) {
+ final StateDirectory stateDirectory,
+ final boolean eosEnabled) {
this.changelogReader = changelogReader;
this.processId = processId;
this.logPrefix = logPrefix;
@@ -103,6 +107,7 @@ public class TaskManager {
this.builder = builder;
this.adminClient = adminClient;
this.stateDirectory = stateDirectory;
+ this.eosEnabled = eosEnabled;
final LogContext logContext = new LogContext(logPrefix);
log = logContext.logger(getClass());
@@ -154,6 +159,7 @@ void handleCorruption(final Map> taskWithChan
final Collection corruptedPartitions = entry.getValue();
task.markChangelogAsCorrupted(corruptedPartitions);
+ task.prepareCloseDirty();
task.closeDirty();
task.revive();
}
@@ -179,6 +185,11 @@ public void handleAssignment(final Map> activeTasks,
// first rectify all existing tasks
final LinkedHashMap taskCloseExceptions = new LinkedHashMap<>();
+
+ final Map> checkpointPerTask = new HashMap<>();
+ final Map> consumedOffsetsAndMetadataPerTask = new HashMap<>();
+ final Set dirtyTasks = new HashSet<>();
+
final Iterator iterator = tasks.values().iterator();
while (iterator.hasNext()) {
final Task task = iterator.next();
@@ -192,30 +203,52 @@ public void handleAssignment(final Map> activeTasks,
cleanupTask(task);
try {
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ final Map committableOffsets = task.committableOffsetsAndMetadata();
+
+ checkpointPerTask.put(task, checkpoint);
+ if (!committableOffsets.isEmpty()) {
+ consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
+ }
} catch (final RuntimeException e) {
final String uncleanMessage = String.format("Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:", task.id());
log.error(uncleanMessage, e);
taskCloseExceptions.put(task.id(), e);
// We've already recorded the exception (which is the point of clean).
// Now, we should go ahead and complete the close because a half-closed task is no good to anyone.
- task.closeDirty();
- } finally {
- if (task.isActive()) {
- try {
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
- } catch (final RuntimeException e) {
- final String uncleanMessage = String.format("Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:", task.id());
- log.error(uncleanMessage, e);
- taskCloseExceptions.putIfAbsent(task.id(), e);
- }
- }
+ task.prepareCloseDirty();
+ dirtyTasks.add(task);
}
iterator.remove();
}
}
+ if (!consumedOffsetsAndMetadataPerTask.isEmpty()) {
+ commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
+ }
+
+ for (final Map.Entry> taskAndCheckpoint : checkpointPerTask.entrySet()) {
+ final Task task = taskAndCheckpoint.getKey();
+ try {
+ task.closeClean(checkpointPerTask.get(task));
+ } catch (final RuntimeException e) {
+ final String uncleanMessage = String.format("Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:", task.id());
+ log.error(uncleanMessage, e);
+ taskCloseExceptions.put(task.id(), e);
+ // We've already recorded the exception (which is the point of clean).
+ // Now, we should go ahead and complete the close because a half-closed task is no good to anyone.
+ task.closeDirty();
+ } finally {
+ cleanUpTaskProducer(task, taskCloseExceptions);
+ }
+ }
+
+ for (final Task task : dirtyTasks) {
+ task.closeDirty();
+ cleanUpTaskProducer(task, taskCloseExceptions);
+ }
+
if (!taskCloseExceptions.isEmpty()) {
for (final Map.Entry entry : taskCloseExceptions.entrySet()) {
if (!(entry.getValue() instanceof TaskMigratedException)) {
@@ -257,6 +290,19 @@ public void handleAssignment(final Map> activeTasks,
changelogReader.transitToRestoreActive();
}
+ private void cleanUpTaskProducer(final Task task,
+ final Map taskCloseExceptions) {
+ if (task.isActive()) {
+ try {
+ activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
+ } catch (final RuntimeException e) {
+ final String uncleanMessage = String.format("Failed to close task %s cleanly. Attempting to close remaining tasks before re-throwing:", task.id());
+ log.error(uncleanMessage, e);
+ taskCloseExceptions.putIfAbsent(task.id(), e);
+ }
+ }
+ }
+
private void addNewTask(final Task task) {
final Task previous = tasks.put(task.id(), task);
if (previous != null) {
@@ -330,13 +376,28 @@ boolean tryToCompleteRestoration() {
void handleRevocation(final Collection revokedPartitions) {
final Set remainingPartitions = new HashSet<>(revokedPartitions);
+ final Map> consumedOffsetsAndMetadataPerTask = new HashMap<>();
for (final Task task : tasks.values()) {
if (remainingPartitions.containsAll(task.inputPartitions())) {
- task.suspend();
+ task.prepareSuspend();
+ final Map committableOffsets = task.committableOffsetsAndMetadata();
+ if (!committableOffsets.isEmpty()) {
+ consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
+ }
}
remainingPartitions.removeAll(task.inputPartitions());
}
+ if (!consumedOffsetsAndMetadataPerTask.isEmpty()) {
+ commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
+ }
+
+ for (final Task task : tasks.values()) {
+ if (consumedOffsetsAndMetadataPerTask.containsKey(task.id())) {
+ task.suspend();
+ }
+ }
+
if (!remainingPartitions.isEmpty()) {
log.warn("The following partitions {} are missing from the task partitions. It could potentially " +
"due to race condition of consumer detecting the heartbeat failure, or the tasks " +
@@ -362,6 +423,7 @@ void handleLostAll() {
// standby tasks while we rejoin.
if (task.isActive()) {
cleanupTask(task);
+ task.prepareCloseDirty();
task.closeDirty();
iterator.remove();
try {
@@ -497,24 +559,52 @@ private void cleanupTask(final Task task) {
void shutdown(final boolean clean) {
final AtomicReference firstException = new AtomicReference<>(null);
- final Iterator iterator = tasks.values().iterator();
- while (iterator.hasNext()) {
- final Task task = iterator.next();
+
+ final Map> checkpointPerTask = new HashMap<>();
+ final Map> consumedOffsetsAndMetadataPerTask = new HashMap<>();
+
+ for (final Task task : tasks.values()) {
cleanupTask(task);
if (clean) {
try {
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ final Map committableOffsets = task.committableOffsetsAndMetadata();
+
+ checkpointPerTask.put(task, checkpoint);
+ if (!committableOffsets.isEmpty()) {
+ consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
+ }
} catch (final TaskMigratedException e) {
// just ignore the exception as it doesn't matter during shutdown
+ task.prepareCloseDirty();
task.closeDirty();
} catch (final RuntimeException e) {
firstException.compareAndSet(null, e);
+ task.prepareCloseDirty();
task.closeDirty();
}
} else {
+ task.prepareCloseDirty();
task.closeDirty();
}
+ }
+
+ if (clean && !consumedOffsetsAndMetadataPerTask.isEmpty()) {
+ commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
+ }
+
+ for (final Map.Entry> taskAndCheckpoint : checkpointPerTask.entrySet()) {
+ final Task task = taskAndCheckpoint.getKey();
+ try {
+ task.closeClean(checkpointPerTask.get(task));
+ } catch (final RuntimeException e) {
+ firstException.compareAndSet(null, e);
+ task.closeDirty();
+ }
+ }
+
+ for (final Task task : tasks.values()) {
if (task.isActive()) {
try {
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
@@ -526,9 +616,10 @@ void shutdown(final boolean clean) {
}
}
}
- iterator.remove();
}
+ tasks.clear();
+
try {
activeTaskCreator.closeThreadProducerIfNeeded();
} catch (final RuntimeException e) {
@@ -604,14 +695,28 @@ int commitAll() {
if (rebalanceInProgress) {
return -1;
} else {
- int commits = 0;
+ int committed = 0;
+ final Map> consumedOffsetsAndMetadataPerTask = new HashMap<>();
+ for (final Task task : tasks.values()) {
+ if (task.commitNeeded()) {
+ task.prepareCommit();
+ final Map offsetAndMetadata = task.committableOffsetsAndMetadata();
+ if (!offsetAndMetadata.isEmpty()) {
+ consumedOffsetsAndMetadataPerTask.put(task.id(), offsetAndMetadata);
+ }
+ }
+ }
+
+ commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
+
for (final Task task : tasks.values()) {
if (task.commitNeeded()) {
- task.commit();
- commits++;
+ ++committed;
+ task.postCommit();
}
}
- return commits;
+
+ return committed;
}
}
@@ -623,14 +728,49 @@ int maybeCommitActiveTasksPerUserRequested() {
if (rebalanceInProgress) {
return -1;
} else {
- int commits = 0;
+ final Map> consumedOffsetsAndMetadataPerTask = new HashMap<>();
for (final Task task : activeTaskIterable()) {
if (task.commitRequested() && task.commitNeeded()) {
- task.commit();
- commits++;
+ task.prepareCommit();
+ final Map offsetAndMetadata = task.committableOffsetsAndMetadata();
+ if (!offsetAndMetadata.isEmpty()) {
+ consumedOffsetsAndMetadataPerTask.put(task.id(), offsetAndMetadata);
+ }
}
}
- return commits;
+
+ commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
+
+ for (final Task task : tasks.values()) {
+ if (consumedOffsetsAndMetadataPerTask.containsKey(task.id())) {
+ task.postCommit();
+ }
+ }
+
+ return consumedOffsetsAndMetadataPerTask.size();
+ }
+ }
+
+ private void commitOffsetsOrTransaction(final Map> offsetsPerTask) {
+ if (eosEnabled) {
+ for (final Map.Entry> taskToCommit : offsetsPerTask.entrySet()) {
+ activeTaskCreator.streamsProducerForTask(taskToCommit.getKey()).commitTransaction(taskToCommit.getValue());
+ }
+ } else {
+ try {
+ final Map allOffsets = offsetsPerTask.values().stream()
+ .flatMap(e -> e.entrySet().stream()).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ mainConsumer.commitSync(allOffsets);
+ } catch (final CommitFailedException error) {
+ throw new TaskMigratedException("Consumer committing offsets failed, " +
+ "indicating the corresponding thread is no longer part of the group", error);
+ } catch (final TimeoutException error) {
+ // TODO KIP-447: we can consider treating it as non-fatal and retry on the thread level
+ throw new StreamsException("Timed out while committing offsets via consumer", error);
+ } catch (final KafkaException error) {
+ throw new StreamsException("Error encountered committing offsets via consumer", error);
+ }
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/MetricsIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/MetricsIntegrationTest.java
index 63ce9c981cb1b..2937498edcb08 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/MetricsIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/MetricsIntegrationTest.java
@@ -61,7 +61,6 @@
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
-@SuppressWarnings("unchecked")
@Category({IntegrationTest.class})
public class MetricsIntegrationTest {
@@ -261,7 +260,7 @@ private void startApplication() throws InterruptedException {
verifyStateMetric(State.CREATED);
verifyTopologyDescriptionMetric(topology.describe().toString());
- verifyApplicationIdMetric(APPLICATION_ID_VALUE);
+ verifyApplicationIdMetric();
kafkaStreams.start();
TestUtils.waitForCondition(
@@ -501,13 +500,13 @@ private void verifyTopologyDescriptionMetric(final String topologyDescription) {
assertThat(metricsList.get(0).metricValue(), is(topologyDescription));
}
- private void verifyApplicationIdMetric(final String applicationId) {
+ private void verifyApplicationIdMetric() {
final List metricsList = new ArrayList(kafkaStreams.metrics().values()).stream()
.filter(m -> m.metricName().name().equals(APPLICATION_ID) &&
m.metricName().group().equals(STREAM_CLIENT_NODE_METRICS))
.collect(Collectors.toList());
assertThat(metricsList.size(), is(1));
- assertThat(metricsList.get(0).metricValue(), is(applicationId));
+ assertThat(metricsList.get(0).metricValue(), is(APPLICATION_ID_VALUE));
}
private void checkClientLevelMetrics() {
@@ -565,10 +564,6 @@ private void checkTaskLevelMetrics(final String builtInMetricsVersion) {
.collect(Collectors.toList());
final int numberOfAddedMetrics = StreamsConfig.METRICS_0100_TO_24.equals(builtInMetricsVersion) ? 0 : 4;
final int numberOfMetricsWithRemovedParent = StreamsConfig.METRICS_0100_TO_24.equals(builtInMetricsVersion) ? 5 : 4;
- checkMetricByName(listMetricTask, COMMIT_LATENCY_AVG, numberOfMetricsWithRemovedParent);
- checkMetricByName(listMetricTask, COMMIT_LATENCY_MAX, numberOfMetricsWithRemovedParent);
- checkMetricByName(listMetricTask, COMMIT_RATE, numberOfMetricsWithRemovedParent);
- checkMetricByName(listMetricTask, COMMIT_TOTAL, numberOfMetricsWithRemovedParent);
checkMetricByName(listMetricTask, ENFORCED_PROCESSING_RATE, 4);
checkMetricByName(listMetricTask, ENFORCED_PROCESSING_TOTAL, 4);
checkMetricByName(listMetricTask, RECORD_LATENESS_AVG, 4);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreatorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreatorTest.java
index 0f9ecdf50043e..b1c4b2c08d73b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreatorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreatorTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.processor.internals;
import java.io.File;
+
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.MockConsumer;
@@ -43,14 +44,22 @@
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.apache.kafka.common.utils.Utils.mkSet;
import static org.easymock.EasyMock.anyString;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.same;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsNot.not;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
@RunWith(EasyMockRunner.class)
public class ActiveTaskCreatorTest {
@@ -73,6 +82,122 @@ public class ActiveTaskCreatorTest {
private ActiveTaskCreator activeTaskCreator;
+ @Test
+ public void shouldFailForNonEosOnStreamsProducerPerTask() {
+ expect(config.getString(StreamsConfig.APPLICATION_ID_CONFIG)).andReturn("appId");
+ expect(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)).andReturn(StreamsConfig.AT_LEAST_ONCE);
+ expect(config.getProducerConfigs(anyString())).andReturn(Collections.emptyMap());
+ replay(config);
+
+ activeTaskCreator = new ActiveTaskCreator(
+ builder,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ changeLogReader,
+ new ThreadCache(new LogContext(), 0L, streamsMetrics),
+ new MockTime(),
+ mockClientSupplier,
+ "threadId",
+ new LogContext().logger(ActiveTaskCreator.class)
+ );
+
+ final IllegalStateException thrown = assertThrows(
+ IllegalStateException.class,
+ () -> activeTaskCreator.streamsProducerForTask(null)
+ );
+
+ assertThat(thrown.getMessage(), is("Producer per thread is used"));
+ }
+
+ @Test
+ public void shouldFailForUnknownTaskOnStreamsProducerPerTask() {
+ expect(config.getString(StreamsConfig.APPLICATION_ID_CONFIG)).andReturn("appId");
+ expect(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)).andReturn(StreamsConfig.EXACTLY_ONCE);
+ expect(config.getProducerConfigs(anyString())).andReturn(Collections.emptyMap());
+ replay(config);
+
+ activeTaskCreator = new ActiveTaskCreator(
+ builder,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ changeLogReader,
+ new ThreadCache(new LogContext(), 0L, streamsMetrics),
+ new MockTime(),
+ mockClientSupplier,
+ "threadId",
+ new LogContext().logger(ActiveTaskCreator.class)
+ );
+
+ {
+ final IllegalStateException thrown = assertThrows(
+ IllegalStateException.class,
+ () -> activeTaskCreator.streamsProducerForTask(null)
+ );
+
+ assertThat(thrown.getMessage(), is("Unknown TaskId: null"));
+ }
+ {
+ final IllegalStateException thrown = assertThrows(
+ IllegalStateException.class,
+ () -> activeTaskCreator.streamsProducerForTask(new TaskId(0, 0))
+ );
+
+ assertThat(thrown.getMessage(), is("Unknown TaskId: 0_0"));
+ }
+ }
+
+ @Test
+ public void shouldReturnStreamsProducerPerTask() {
+ final TaskId task00 = new TaskId(0, 0);
+ final TaskId task01 = new TaskId(0, 1);
+ final ProcessorTopology topology = mock(ProcessorTopology.class);
+
+ expect(config.getString(StreamsConfig.APPLICATION_ID_CONFIG)).andReturn("appId");
+ expect(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)).andReturn(StreamsConfig.EXACTLY_ONCE);
+ expect(config.getProducerConfigs(anyString())).andReturn(new HashMap<>()).anyTimes();
+ expect(config.getLong(anyString())).andReturn(0L).anyTimes();
+ expect(config.getInt(anyString())).andReturn(0).anyTimes();
+ expect(builder.buildSubtopology(task00.topicGroupId)).andReturn(topology).anyTimes();
+ expect(stateDirectory.directoryForTask(task00)).andReturn(new File(task00.toString()));
+ expect(stateDirectory.directoryForTask(task01)).andReturn(new File(task01.toString()));
+ expect(topology.storeToChangelogTopic()).andReturn(Collections.emptyMap()).anyTimes();
+ expect(topology.source("topic")).andReturn(mock(SourceNode.class)).andReturn(mock(SourceNode.class));
+ expect(topology.globalStateStores()).andReturn(Collections.emptyList()).anyTimes();
+ replay(config, builder, stateDirectory, topology);
+
+ mockClientSupplier.setApplicationIdForProducer("appId");
+ activeTaskCreator = new ActiveTaskCreator(
+ builder,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ changeLogReader,
+ new ThreadCache(new LogContext(), 0L, streamsMetrics),
+ new MockTime(),
+ mockClientSupplier,
+ "threadId",
+ new LogContext().logger(ActiveTaskCreator.class)
+ );
+
+ assertThat(
+ activeTaskCreator.createTasks(
+ null,
+ mkMap(
+ mkEntry(task00, Collections.singleton(new TopicPartition("topic", 0))),
+ mkEntry(task01, Collections.singleton(new TopicPartition("topic", 1)))
+ )
+ ).stream().map(Task::id).collect(Collectors.toSet()),
+ equalTo(mkSet(task00, task01))
+ );
+
+ final StreamsProducer streamsProducer1 = activeTaskCreator.streamsProducerForTask(new TaskId(0, 0));
+ final StreamsProducer streamsProducer2 = activeTaskCreator.streamsProducerForTask(new TaskId(0, 1));
+
+ assertThat(streamsProducer1, not(same(streamsProducer2)));
+ }
+
@Test
public void shouldConstructProducerMetricsWithoutEOS() {
expect(config.getString(StreamsConfig.APPLICATION_ID_CONFIG)).andReturn("appId");
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
index e09d99c952cab..5df3f12780978 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
@@ -16,11 +16,6 @@
*/
package org.apache.kafka.streams.processor.internals;
-import org.apache.kafka.clients.consumer.CommitFailedException;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -35,7 +30,6 @@
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.ProducerFencedException;
-import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
@@ -67,6 +61,7 @@
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.replay;
@@ -104,10 +99,9 @@ public class RecordCollectorTest {
private final StreamPartitioner streamPartitioner = (topic, key, value, numPartitions) -> Integer.parseInt(key) % numPartitions;
- private final MockConsumer mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
private final MockProducer mockProducer = new MockProducer<>(
cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer);
- private final StreamsProducer streamsProducer = new StreamsProducer(mockProducer, false, logContext, null);
+ private final StreamsProducer streamsProducer = new StreamsProducer(mockProducer, false, null, logContext);
private RecordCollectorImpl collector;
@@ -116,10 +110,8 @@ public void setup() {
collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
streamsProducer,
productionExceptionHandler,
- false,
streamsMetrics);
}
@@ -243,51 +235,29 @@ public void shouldPassThroughRecordHeaderToSerializer() {
}
@Test
- public void shouldCommitViaConsumerIfEosDisabled() {
- final KafkaConsumer consumer = mock(KafkaConsumer.class);
- consumer.commitSync((Map) null);
- expectLastCall();
- replay(consumer);
-
- final RecordCollector collector = new RecordCollectorImpl(
- logContext,
- taskId,
- consumer,
- streamsProducer,
- productionExceptionHandler,
- false,
- streamsMetrics);
-
- collector.commit(null);
-
- verify(consumer);
-
- }
-
- @Test
- public void shouldCommitViaProducerIfEosEnabled() {
+ public void shouldForwardFlushToStreamsProducer() {
final StreamsProducer streamsProducer = mock(StreamsProducer.class);
- streamsProducer.commitTransaction(null);
+ expect(streamsProducer.eosEnabled()).andReturn(false);
+ streamsProducer.flush();
expectLastCall();
replay(streamsProducer);
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
streamsProducer,
productionExceptionHandler,
- true,
streamsMetrics);
- collector.commit(null);
+ collector.flush();
verify(streamsProducer);
}
@Test
- public void shouldForwardFlushToTransactionManager() {
+ public void shouldForwardFlushToStreamsProducerEosEnabled() {
final StreamsProducer streamsProducer = mock(StreamsProducer.class);
+ expect(streamsProducer.eosEnabled()).andReturn(true);
streamsProducer.flush();
expectLastCall();
replay(streamsProducer);
@@ -295,10 +265,8 @@ public void shouldForwardFlushToTransactionManager() {
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
streamsProducer,
productionExceptionHandler,
- true,
streamsMetrics);
collector.flush();
@@ -306,38 +274,18 @@ public void shouldForwardFlushToTransactionManager() {
verify(streamsProducer);
}
- @Test
- public void shouldForwardCloseToTransactionManager() {
- final StreamsProducer streamsProducer = mock(StreamsProducer.class);
- replay(streamsProducer);
-
- final RecordCollector collector = new RecordCollectorImpl(
- logContext,
- taskId,
- mockConsumer,
- streamsProducer,
- productionExceptionHandler,
- false,
- streamsMetrics);
-
- collector.close();
-
- verify(streamsProducer);
- }
-
@Test
public void shouldAbortTxIfEosEnabled() {
final StreamsProducer streamsProducer = mock(StreamsProducer.class);
+ expect(streamsProducer.eosEnabled()).andReturn(true);
streamsProducer.abortTransaction();
replay(streamsProducer);
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
streamsProducer,
productionExceptionHandler,
- true,
streamsMetrics);
collector.close();
@@ -345,7 +293,7 @@ public void shouldAbortTxIfEosEnabled() {
verify(streamsProducer);
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
@Test
public void shouldThrowInformativeStreamsExceptionOnKeyClassCastException() {
final StreamsException expected = assertThrows(
@@ -373,7 +321,7 @@ public void shouldThrowInformativeStreamsExceptionOnKeyClassCastException() {
);
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
@Test
public void shouldThrowInformativeStreamsExceptionOnKeyAndNullValueClassCastException() {
final StreamsException expected = assertThrows(
@@ -401,7 +349,7 @@ public void shouldThrowInformativeStreamsExceptionOnKeyAndNullValueClassCastExce
);
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
@Test
public void shouldThrowInformativeStreamsExceptionOnValueClassCastException() {
final StreamsException expected = assertThrows(
@@ -429,7 +377,7 @@ public void shouldThrowInformativeStreamsExceptionOnValueClassCastException() {
);
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
@Test
public void shouldThrowInformativeStreamsExceptionOnValueAndNullKeyClassCastException() {
final StreamsException expected = assertThrows(
@@ -463,7 +411,6 @@ public void shouldThrowTaskMigratedExceptionOnSubsequentCallWhenProducerFencedIn
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
new StreamsProducer(
new MockProducer(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer) {
@Override
@@ -473,11 +420,10 @@ public synchronized Future send(final ProducerRecord(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer) {
@Override
@@ -516,11 +461,10 @@ public synchronized Future send(final ProducerRecord(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer) {
@Override
@@ -558,11 +501,10 @@ public synchronized Future send(final ProducerRecord(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer) {
@Override
@@ -601,11 +542,10 @@ public synchronized Future send(final ProducerRecord send(final ProducerRecord(OffsetResetStrategy.EARLIEST) {
- @Override
- public void commitSync(final Map offsets) {
- throw new CommitFailedException();
- }
- },
- streamsProducer,
- productionExceptionHandler,
- false,
- streamsMetrics
- );
-
- final TaskMigratedException thrown = assertThrows(TaskMigratedException.class, () -> collector.commit(null));
-
- assertThat(thrown.getMessage(), equalTo("Consumer committing offsets failed, indicating the corresponding thread is no longer part of the group; it means all tasks belonging to this thread should be migrated."));
- }
-
- @Test
- public void shouldThrowStreamsExceptionOnCommitTimeout() {
- final RecordCollector collector = new RecordCollectorImpl(
- logContext,
- taskId,
- new MockConsumer(OffsetResetStrategy.EARLIEST) {
- @Override
- public void commitSync(final Map offsets) {
- throw new TimeoutException();
- }
- },
- streamsProducer,
- productionExceptionHandler,
- false,
- streamsMetrics
- );
-
- final StreamsException thrown = assertThrows(StreamsException.class, () -> collector.commit(null));
-
- assertThat(thrown.getMessage(), equalTo("Timed out while committing offsets via consumer for task 0_0"));
- }
-
- @Test
- public void shouldStreamsExceptionOnCommitError() {
- final RecordCollector collector = new RecordCollectorImpl(
- logContext,
- taskId,
- new MockConsumer(OffsetResetStrategy.EARLIEST) {
- @Override
- public void commitSync(final Map offsets) {
- throw new KafkaException();
- }
- },
- streamsProducer,
- productionExceptionHandler,
- false,
- streamsMetrics
- );
- collector.initialize();
-
- final StreamsException thrown = assertThrows(StreamsException.class, () -> collector.commit(null));
-
- assertThat(thrown.getMessage(), equalTo("Error encountered committing offsets via consumer for task 0_0"));
- }
-
- @Test
- public void shouldFailOnCommitFatal() {
- final RecordCollector collector = new RecordCollectorImpl(
- logContext,
- taskId,
- new MockConsumer(OffsetResetStrategy.EARLIEST) {
- @Override
- public void commitSync(final Map offsets) {
- throw new RuntimeException("KABOOM!");
- }
- },
- streamsProducer,
- productionExceptionHandler,
- false,
- streamsMetrics
- );
- collector.initialize();
-
- final RuntimeException thrown = assertThrows(RuntimeException.class, () -> collector.commit(null));
-
- assertThat(thrown.getMessage(), equalTo("KABOOM!"));
- }
-
@Test
public void shouldNotAbortTxnOnEOSCloseIfNothingSent() {
final AtomicBoolean functionCalled = new AtomicBoolean(false);
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
new StreamsProducer(
new MockProducer(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer) {
@Override
@@ -732,11 +581,10 @@ public void abortTransaction() {
}
},
true,
- logContext,
- "appId"
+ "appId",
+ logContext
),
productionExceptionHandler,
- true,
streamsMetrics
);
@@ -749,7 +597,6 @@ public void shouldThrowIfTopicIsUnknownOnSendWithPartitioner() {
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
new StreamsProducer(
new MockProducer(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer) {
@Override
@@ -758,11 +605,10 @@ public List partitionsFor(final String topic) {
}
},
false,
- logContext,
- null
+ null,
+ logContext
),
productionExceptionHandler,
- false,
streamsMetrics
);
collector.initialize();
@@ -779,10 +625,8 @@ public void shouldNotCloseInternalProducerForEOS() {
final RecordCollector collector = new RecordCollectorImpl(
logContext,
taskId,
- mockConsumer,
- new StreamsProducer(mockProducer, true, logContext, "appId"),
+ new StreamsProducer(mockProducer, true, "appId", logContext),
productionExceptionHandler,
- true,
streamsMetrics
);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
index 2bde646e7b8d1..a12da5b45d311 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
@@ -49,6 +49,7 @@
import java.io.File;
import java.io.IOException;
import java.util.Collections;
+import java.util.Map;
import static java.util.Arrays.asList;
import static org.apache.kafka.common.utils.Utils.mkEntry;
@@ -133,6 +134,7 @@ public void setup() throws Exception {
@After
public void cleanup() throws IOException {
if (task != null && !task.isClosed()) {
+ task.prepareCloseDirty();
task.closeDirty();
task = null;
}
@@ -179,7 +181,7 @@ public void shouldTransitToRunningAfterInitialization() {
public void shouldThrowIfCommittingOnIllegalState() {
task = createStandbyTask();
- assertThrows(IllegalStateException.class, task::commit);
+ assertThrows(IllegalStateException.class, task::prepareCommit);
}
@Test
@@ -193,7 +195,8 @@ public void shouldFlushAndCheckpointStateManagerOnCommit() {
task = createStandbyTask();
task.initializeIfNeeded();
- task.commit();
+ task.prepareCommit();
+ task.postCommit();
EasyMock.verify(stateManager);
}
@@ -222,7 +225,8 @@ public void shouldDoNothingWithCreatedStateOnClose() {
final MetricName metricName = setupCloseTaskMetric();
task = createStandbyTask();
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
assertEquals(Task.State.CLOSED, task.state());
@@ -246,6 +250,7 @@ public void shouldNotCommitAndThrowOnCloseDirty() {
task = createStandbyTask();
task.initializeIfNeeded();
+ task.prepareCloseDirty();
task.closeDirty();
assertEquals(Task.State.CLOSED, task.state());
@@ -285,7 +290,8 @@ public void shouldCommitOnCloseClean() {
task = createStandbyTask();
task.initializeIfNeeded();
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
assertEquals(Task.State.CLOSED, task.state());
@@ -313,7 +319,8 @@ public void shouldOnlyNeedCommitWhenChangelogOffsetChanged() {
assertTrue(task.commitNeeded());
- task.commit();
+ task.prepareCommit();
+ task.postCommit();
// do not need to commit if there's no update
assertFalse(task.commitNeeded());
@@ -336,9 +343,8 @@ public void shouldThrowOnCloseCleanError() {
task = createStandbyTask();
task.initializeIfNeeded();
- assertThrows(RuntimeException.class, task::closeClean);
-
- assertEquals(Task.State.CLOSING, task.state());
+ final Map checkpoint = task.prepareCloseClean();
+ assertThrows(RuntimeException.class, () -> task.closeClean(checkpoint));
final double expectedCloseTaskMetric = 0.0;
verifyCloseTaskMetric(expectedCloseTaskMetric, streamsMetrics, metricName);
@@ -360,7 +366,7 @@ public void shouldThrowOnCloseCleanFlushError() {
task = createStandbyTask();
task.initializeIfNeeded();
- assertThrows(RuntimeException.class, task::closeClean);
+ assertThrows(RuntimeException.class, task::prepareCloseClean);
assertEquals(Task.State.RUNNING, task.state());
final double expectedCloseTaskMetric = 0.0;
@@ -383,7 +389,8 @@ public void shouldThrowOnCloseCleanCheckpointError() {
task = createStandbyTask();
task.initializeIfNeeded();
- assertThrows(RuntimeException.class, task::closeClean);
+ task.prepareCommit();
+ assertThrows(RuntimeException.class, task::postCommit);
assertEquals(Task.State.RUNNING, task.state());
@@ -400,11 +407,12 @@ public void shouldThrowOnCloseCleanCheckpointError() {
public void shouldThrowIfClosingOnIllegalState() {
task = createStandbyTask();
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
// close call are not idempotent since we are already in closed
- assertThrows(IllegalStateException.class, task::closeClean);
- assertThrows(IllegalStateException.class, task::closeDirty);
+ assertThrows(IllegalStateException.class, task::prepareCloseClean);
+ assertThrows(IllegalStateException.class, task::prepareCloseDirty);
}
private StandbyTask createStandbyTask() {
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index dc0404b16ba07..5218020e29986 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -86,6 +86,7 @@
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -207,6 +208,7 @@ public void setup() {
@After
public void cleanup() throws IOException {
if (task != null && !task.isClosed()) {
+ task.prepareCloseDirty();
task.closeDirty();
task = null;
}
@@ -264,7 +266,7 @@ public void shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() thr
ctrl.replay();
task = createStatefulTask(createConfig(true, "100"), true, stateManager);
- task.transitionTo(Task.State.CLOSING);
+ task.prepareCloseDirty();
task.closeDirty();
task = null;
@@ -382,31 +384,6 @@ public void shouldConstructMetricsWithBuiltInMetricsVersionLatest() {
private void testMetrics(final String builtInMetricsVersion) {
task = createStatelessTask(createConfig(false, "100"), builtInMetricsVersion);
- assertNotNull(getMetric(
- "commit",
- "%s-latency-avg",
- task.id().toString(),
- builtInMetricsVersion
- ));
- assertNotNull(getMetric(
- "commit",
- "%s-latency-max",
- task.id().toString(),
- builtInMetricsVersion
- ));
- assertNotNull(getMetric(
- "commit",
- "%s-rate",
- task.id().toString(),
- builtInMetricsVersion
- ));
- assertNotNull(getMetric(
- "commit",
- "%s-total",
- task.id().toString(),
- builtInMetricsVersion
- ));
-
assertNotNull(getMetric(
"enforced-processing",
"%s-rate",
@@ -714,44 +691,80 @@ public void shouldRespectCommitNeeded() {
assertTrue(task.process(0L));
assertTrue(task.commitNeeded());
- task.commit();
+ task.prepareCommit();
+ assertTrue(task.commitNeeded());
+
+ task.postCommit();
assertFalse(task.commitNeeded());
assertTrue(task.maybePunctuateStreamTime());
assertTrue(task.commitNeeded());
- task.commit();
+ task.prepareCommit();
+ assertTrue(task.commitNeeded());
+
+ task.postCommit();
assertFalse(task.commitNeeded());
time.sleep(10);
assertTrue(task.maybePunctuateSystemTime());
assertTrue(task.commitNeeded());
- task.commit();
+ task.prepareCommit();
+ assertTrue(task.commitNeeded());
+
+ task.postCommit();
assertFalse(task.commitNeeded());
}
@Test
public void shouldCommitNextOffsetFromQueueIfAvailable() {
- recordCollector.commit(EasyMock.eq(mkMap(mkEntry(partition1, new OffsetAndMetadata(5L, encodeTimestamp(5L))))));
- EasyMock.expectLastCall();
-
task = createStatelessTask(createConfig(false, "0"), StreamsConfig.METRICS_LATEST);
task.initializeIfNeeded();
task.completeRestoration();
task.addRecords(partition1, Arrays.asList(getConsumerRecord(partition1, 0L), getConsumerRecord(partition1, 5L)));
task.process(0L);
- task.commit();
+ task.prepareCommit();
+ final Map offsetsAndMetadata = task.committableOffsetsAndMetadata();
- EasyMock.verify(recordCollector);
+ assertThat(offsetsAndMetadata, equalTo(mkMap(mkEntry(partition1, new OffsetAndMetadata(5L, encodeTimestamp(5L))))));
}
@Test
public void shouldCommitConsumerPositionIfRecordQueueIsEmpty() {
- recordCollector.commit(EasyMock.eq(mkMap(mkEntry(partition1, new OffsetAndMetadata(3L, encodeTimestamp(0L))))));
- EasyMock.expectLastCall();
+ task = createStatelessTask(createConfig(false, "0"), StreamsConfig.METRICS_LATEST);
+ task.initializeIfNeeded();
+ task.completeRestoration();
+
+ consumer.addRecord(getConsumerRecord(partition1, 0L));
+ consumer.addRecord(getConsumerRecord(partition1, 1L));
+ consumer.addRecord(getConsumerRecord(partition1, 2L));
+ consumer.poll(Duration.ZERO);
+
+ task.addRecords(partition1, singletonList(getConsumerRecord(partition1, 0L)));
+ task.process(0L);
+ task.prepareCommit();
+ final Map offsetsAndMetadata = task.committableOffsetsAndMetadata();
+
+ assertThat(offsetsAndMetadata, equalTo(mkMap(mkEntry(partition1, new OffsetAndMetadata(3L, encodeTimestamp(0L))))));
+ }
+
+ @Test
+ public void shouldFailOnCommitIfTaskIsClosed() {
+ task = createStatelessTask(createConfig(false, "0"), StreamsConfig.METRICS_LATEST);
+ task.transitionTo(Task.State.CLOSED);
+
+ final IllegalStateException thrown = assertThrows(
+ IllegalStateException.class,
+ task::committableOffsetsAndMetadata
+ );
+
+ assertThat(thrown.getMessage(), is("Task 0_0 is closed."));
+ }
+ @Test
+ public void shouldOnlyCommitConsumerPositionTaskIfRunning() {
task = createStatelessTask(createConfig(false, "0"), StreamsConfig.METRICS_LATEST);
task.initializeIfNeeded();
task.completeRestoration();
@@ -763,9 +776,23 @@ public void shouldCommitConsumerPositionIfRecordQueueIsEmpty() {
task.addRecords(partition1, singletonList(getConsumerRecord(partition1, 0L)));
task.process(0L);
- task.commit();
+ task.prepareCommit();
+ task.postCommit();
+
+ final Map offsetsAndMetadata = task.committableOffsetsAndMetadata();
+
+ assertThat(offsetsAndMetadata, equalTo(mkMap(mkEntry(partition1, new OffsetAndMetadata(3L, encodeTimestamp(0L))))));
+
+ task.transitionTo(Task.State.SUSPENDED);
+ assertTrue(task.committableOffsetsAndMetadata().isEmpty());
+
+ task.transitionTo(Task.State.CLOSED);
+ task.transitionTo(Task.State.CREATED);
+ assertTrue(task.committableOffsetsAndMetadata().isEmpty());
+
+ task.transitionTo(Task.State.RESTORING);
+ assertTrue(task.committableOffsetsAndMetadata().isEmpty());
- EasyMock.verify(recordCollector);
}
@Test
@@ -1041,8 +1068,6 @@ public void shouldCommitWhenSuspending() throws IOException {
stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
EasyMock.expect(stateDirectory.lock(taskId)).andReturn(true);
EasyMock.expect(recordCollector.offsets()).andReturn(Collections.singletonMap(changelogPartition, 10L));
- recordCollector.commit(EasyMock.eq(Collections.emptyMap()));
- EasyMock.expectLastCall();
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.emptySet()).anyTimes();
stateManager.checkpoint(EasyMock.eq(Collections.singletonMap(changelogPartition, 10L)));
EasyMock.expectLastCall();
@@ -1053,6 +1078,7 @@ public void shouldCommitWhenSuspending() throws IOException {
task.initializeIfNeeded();
task.completeRestoration();
+ task.prepareSuspend();
task.suspend();
assertEquals(Task.State.SUSPENDED, task.state());
@@ -1082,8 +1108,6 @@ public void shouldNotReInitializeTopologyWhenResuming() throws IOException {
stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
EasyMock.expect(stateDirectory.lock(taskId)).andReturn(true);
EasyMock.expect(recordCollector.offsets()).andThrow(new AssertionError("Should not try to read offsets")).anyTimes();
- recordCollector.commit(EasyMock.anyObject());
- EasyMock.expectLastCall().andThrow(new AssertionError("Should not try to commit")).anyTimes();
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.emptySet()).anyTimes();
stateManager.checkpoint(EasyMock.eq(Collections.emptyMap()));
EasyMock.expectLastCall();
@@ -1094,6 +1118,7 @@ public void shouldNotReInitializeTopologyWhenResuming() throws IOException {
task.initializeIfNeeded();
+ task.prepareSuspend();
task.suspend();
assertEquals(Task.State.SUSPENDED, task.state());
@@ -1120,8 +1145,6 @@ public void shouldCheckpointOffsetsOnCommit() {
final Long offset = 543L;
EasyMock.expect(recordCollector.offsets()).andReturn(Collections.singletonMap(changelogPartition, offset));
- recordCollector.commit(EasyMock.eq(Collections.emptyMap()));
- EasyMock.expectLastCall();
stateManager.checkpoint(EasyMock.eq(Collections.singletonMap(changelogPartition, offset)));
EasyMock.expectLastCall();
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.singleton(changelogPartition));
@@ -1133,7 +1156,8 @@ public void shouldCheckpointOffsetsOnCommit() {
task.initializeIfNeeded();
task.completeRestoration();
- task.commit();
+ task.prepareCommit();
+ task.postCommit();
EasyMock.verify(recordCollector);
}
@@ -1149,7 +1173,8 @@ public void shouldNotCheckpointOffsetsOnCommitIfEosIsEnabled() {
task.initializeIfNeeded();
task.completeRestoration();
- task.commit();
+ task.prepareCommit();
+ task.postCommit();
final File checkpointFile = new File(
stateDirectory.directoryForTask(taskId),
StateManagerUtil.CHECKPOINT_FILE_NAME
@@ -1220,6 +1245,7 @@ public void shouldCloseStateManagerEvenFailureOnUncleanTaskClose() {
task.initializeIfNeeded();
task.completeRestoration();
+ task.prepareCloseDirty();
task.closeDirty();
EasyMock.verify(stateManager);
@@ -1262,7 +1288,7 @@ public void shouldReturnOffsetsForRepartitionTopicsForPurging() {
assertTrue(task.process(0L));
assertTrue(task.process(0L));
- task.commit();
+ task.prepareCommit();
final Map map = task.purgeableOffsets();
@@ -1291,8 +1317,19 @@ public Map committed(final Set checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
assertEquals(Task.State.CLOSED, task.state());
assertFalse(source1.initialized);
@@ -1354,6 +1392,7 @@ public void shouldNotCommitAndThrowOnCloseDirty() {
task.initializeIfNeeded();
task.completeRestoration();
+ task.prepareCloseDirty();
task.closeDirty();
assertEquals(Task.State.CLOSED, task.state());
@@ -1372,14 +1411,13 @@ public void shouldNotCommitOnSuspendRestoring() {
EasyMock.expectLastCall();
stateManager.checkpoint(EasyMock.eq(Collections.emptyMap()));
EasyMock.expectLastCall();
- recordCollector.commit(EasyMock.anyObject());
- EasyMock.expectLastCall().andThrow(new AssertionError("Should not call this function")).anyTimes();
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.emptySet()).anyTimes();
EasyMock.replay(stateManager);
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
task.initializeIfNeeded();
+ task.prepareSuspend();
task.suspend();
assertEquals(Task.State.SUSPENDED, task.state());
@@ -1393,15 +1431,14 @@ public void shouldNotCommitOnCloseRestoring() {
EasyMock.expectLastCall();
stateManager.checkpoint(EasyMock.eq(Collections.emptyMap()));
EasyMock.expectLastCall();
- recordCollector.commit(EasyMock.anyObject());
- EasyMock.expectLastCall().andThrow(new AssertionError("Should not call this function")).anyTimes();
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.emptySet()).anyTimes();
EasyMock.replay(stateManager);
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
task.initializeIfNeeded();
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
assertEquals(Task.State.CLOSED, task.state());
@@ -1413,8 +1450,6 @@ public void shouldCommitOnCloseClean() {
final long offset = 543L;
EasyMock.expect(recordCollector.offsets()).andReturn(Collections.singletonMap(changelogPartition, offset));
- recordCollector.commit(EasyMock.eq(Collections.emptyMap()));
- EasyMock.expectLastCall();
stateManager.close();
EasyMock.expectLastCall();
stateManager.flush();
@@ -1428,7 +1463,8 @@ public void shouldCommitOnCloseClean() {
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
task.initializeIfNeeded();
task.completeRestoration();
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
assertEquals(Task.State.CLOSED, task.state());
@@ -1439,12 +1475,10 @@ public void shouldCommitOnCloseClean() {
}
@Test
- public void shouldThrowOnCloseCleanError() {
+ public void shouldSwallowExceptionOnCloseCleanError() {
final long offset = 543L;
EasyMock.expect(recordCollector.offsets()).andReturn(Collections.singletonMap(changelogPartition, offset));
- recordCollector.commit(EasyMock.eq(Collections.emptyMap()));
- EasyMock.expectLastCall();
stateManager.checkpoint(EasyMock.eq(Collections.singletonMap(changelogPartition, offset)));
EasyMock.expectLastCall();
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.singleton(changelogPartition)).anyTimes();
@@ -1457,9 +1491,8 @@ public void shouldThrowOnCloseCleanError() {
task.initializeIfNeeded();
task.completeRestoration();
- assertThrows(ProcessorStateException.class, task::closeClean);
-
- assertEquals(Task.State.CLOSING, task.state());
+ final Map checkpoint = task.prepareCloseClean();
+ assertThrows(ProcessorStateException.class, () -> task.closeClean(checkpoint));
final double expectedCloseTaskMetric = 0.0;
verifyCloseTaskMetric(expectedCloseTaskMetric, streamsMetrics, metricName);
@@ -1467,6 +1500,8 @@ public void shouldThrowOnCloseCleanError() {
EasyMock.verify(stateManager);
EasyMock.reset(stateManager);
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.singleton(changelogPartition)).anyTimes();
+ stateManager.close();
+ EasyMock.expectLastCall();
EasyMock.replay(stateManager);
}
@@ -1475,8 +1510,6 @@ public void shouldThrowOnCloseCleanFlushError() {
final long offset = 543L;
EasyMock.expect(recordCollector.offsets()).andReturn(Collections.singletonMap(changelogPartition, offset));
- recordCollector.commit(EasyMock.eq(Collections.emptyMap()));
- EasyMock.expectLastCall();
stateManager.flush();
EasyMock.expectLastCall().andThrow(new ProcessorStateException("KABOOM!")).anyTimes();
stateManager.checkpoint(EasyMock.anyObject());
@@ -1490,7 +1523,7 @@ public void shouldThrowOnCloseCleanFlushError() {
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
task.initializeIfNeeded();
- assertThrows(ProcessorStateException.class, task::closeClean);
+ assertThrows(ProcessorStateException.class, task::prepareCloseClean);
assertEquals(Task.State.RESTORING, task.state());
@@ -1508,8 +1541,6 @@ public void shouldThrowOnCloseCleanCheckpointError() {
final long offset = 543L;
EasyMock.expect(recordCollector.offsets()).andReturn(Collections.singletonMap(changelogPartition, offset));
- recordCollector.commit(EasyMock.eq(Collections.emptyMap()));
- EasyMock.expectLastCall();
stateManager.flush();
EasyMock.expectLastCall();
stateManager.checkpoint(Collections.emptyMap());
@@ -1524,7 +1555,8 @@ public void shouldThrowOnCloseCleanCheckpointError() {
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
task.initializeIfNeeded();
- assertThrows(ProcessorStateException.class, task::closeClean);
+ final Map checkpoint = task.prepareCloseClean();
+ assertThrows(ProcessorStateException.class, () -> task.closeClean(checkpoint));
assertEquals(Task.State.RESTORING, task.state());
@@ -1558,10 +1590,11 @@ public void shouldThrowIfClosingOnIllegalState() {
task = createOptimizedStatefulTask(createConfig(false, "100"), consumer);
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
// close call are not idempotent since we are already in closed
- assertThrows(IllegalStateException.class, task::closeClean);
+ assertThrows(IllegalStateException.class, () -> task.closeClean(checkpoint));
assertThrows(IllegalStateException.class, task::closeDirty);
EasyMock.reset(stateManager);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index 6cd712945af62..2ddafd0295032 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -1046,6 +1046,10 @@ public void shouldNotCloseTaskAndRemoveFromTaskManagerIfProducerGotFencedInCommi
assertThat(thread.activeTasks().size(), equalTo(1));
+ // need to process a record to enable committing
+ addRecord(mockConsumer, 0L);
+ thread.runOnce();
+
clientSupplier.producers.get(0).commitTransactionException = new ProducerFencedException("Producer is fenced");
assertThrows(TaskMigratedException.class, () -> thread.rebalanceListener.onPartitionsRevoked(assignedPartitions));
assertFalse(clientSupplier.producers.get(0).transactionCommitted());
@@ -1133,6 +1137,10 @@ public void shouldNotCloseTaskProducerWhenSuspending() {
assertThat(thread.activeTasks().size(), equalTo(1));
+ // need to process a record to enable committing
+ addRecord(mockConsumer, 0L);
+ thread.runOnce();
+
thread.rebalanceListener.onPartitionsRevoked(assignedPartitions);
assertTrue(clientSupplier.producers.get(0).transactionCommitted());
assertFalse(clientSupplier.producers.get(0).closed());
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
index c9945c37375e1..90dedab38d918 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
@@ -70,15 +70,15 @@ public class StreamsProducerTest {
mkEntry(new TopicPartition(topic, 0), new OffsetAndMetadata(0L, null))
);
- private final MockProducer mockProducer = new MockProducer<>(
+ private final MockProducer nonEosMockProducer = new MockProducer<>(
cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer);
- private final StreamsProducer aloStreamsProducer =
- new StreamsProducer(mockProducer, false, logContext, null);
+ private final StreamsProducer nonEosStreamsProducer =
+ new StreamsProducer(nonEosMockProducer, false, null, logContext);
private final MockProducer eosMockProducer = new MockProducer<>(
cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer);
private final StreamsProducer eosStreamsProducer =
- new StreamsProducer(eosMockProducer, true, logContext, "appId");
+ new StreamsProducer(eosMockProducer, true, "appId", logContext);
private final ProducerRecord record =
new ProducerRecord<>(topic, 0, 0L, new byte[0], new byte[0], new RecordHeaders());
@@ -88,44 +88,11 @@ public void before() {
eosStreamsProducer.initTransaction();
}
- @Test
- public void shouldFailIfProducerIsNull() {
- {
- final NullPointerException thrown = assertThrows(
- NullPointerException.class,
- () -> new StreamsProducer(null, false, logContext, null)
- );
- assertThat(thrown.getMessage(), is("producer cannot be null"));
- }
-
- {
- final NullPointerException thrown = assertThrows(
- NullPointerException.class,
- () -> new StreamsProducer(null, true, logContext, "appId")
- );
-
- assertThat(thrown.getMessage(), is("producer cannot be null"));
- }
- }
-
- @Test
- public void shouldNotInitTxIfEosDisable() {
- assertThat(mockProducer.transactionInitialized(), is(false));
- }
- @Test
- public void shouldNotBeginTxOnSendIfEosDisable() {
- aloStreamsProducer.send(record, null);
- assertThat(mockProducer.transactionInFlight(), is(false));
- }
+ // generic tests (non-EOS and EOS)
- @Test
- public void shouldForwardRecordOnSend() {
- aloStreamsProducer.send(record, null);
- assertThat(mockProducer.history().size(), is(1));
- assertThat(mockProducer.history().get(0), is(record));
- }
+ // functional tests
@Test
public void shouldForwardCallToPartitionsFor() {
@@ -136,7 +103,7 @@ public void shouldForwardCallToPartitionsFor() {
replay(producer);
final StreamsProducer streamsProducer =
- new StreamsProducer(producer, false, logContext, null);
+ new StreamsProducer(producer, false, null, logContext);
final List partitionInfo = streamsProducer.partitionsFor(topic);
@@ -153,7 +120,7 @@ public void shouldForwardCallToFlush() {
replay(producer);
final StreamsProducer streamsProducer =
- new StreamsProducer(producer, false, logContext, null);
+ new StreamsProducer(producer, false, null, logContext);
streamsProducer.flush();
@@ -162,37 +129,92 @@ public void shouldForwardCallToFlush() {
// error handling tests
+ @Test
+ public void shouldFailIfProducerIsNull() {
+ {
+ final NullPointerException thrown = assertThrows(
+ NullPointerException.class,
+ () -> new StreamsProducer(null, false, "appId", logContext)
+ );
+
+ assertThat(thrown.getMessage(), is("producer cannot be null"));
+ }
+
+ {
+ final NullPointerException thrown = assertThrows(
+ NullPointerException.class,
+ () -> new StreamsProducer(null, true, "appId", logContext)
+ );
+
+ assertThat(thrown.getMessage(), is("producer cannot be null"));
+ }
+ }
+
+ @Test
+ public void shouldFailIfLogContextIsNull() {
+ final NullPointerException thrown = assertThrows(
+ NullPointerException.class,
+ () -> new StreamsProducer(nonEosMockProducer, false, "appId", null)
+ );
+
+ assertThat(thrown.getMessage(), is("logContext cannot be null"));
+ }
+
+
+ // non-EOS tests
+
+ // functional tests
+
+ @Test
+ public void shouldNotInitTxIfEosDisable() {
+ assertThat(nonEosMockProducer.transactionInitialized(), is(false));
+ }
+
+ @Test
+ public void shouldNotBeginTxOnSendIfEosDisable() {
+ nonEosStreamsProducer.send(record, null);
+ assertThat(nonEosMockProducer.transactionInFlight(), is(false));
+ }
+
+ @Test
+ public void shouldForwardRecordOnSend() {
+ nonEosStreamsProducer.send(record, null);
+ assertThat(nonEosMockProducer.history().size(), is(1));
+ assertThat(nonEosMockProducer.history().get(0), is(record));
+ }
+
+ // error handling tests
+
@Test
public void shouldFailOnInitTxIfEosDisabled() {
final IllegalStateException thrown = assertThrows(
IllegalStateException.class,
- aloStreamsProducer::initTransaction
+ nonEosStreamsProducer::initTransaction
);
- assertThat(thrown.getMessage(), is("EOS is disabled [test, alo]"));
+ assertThat(thrown.getMessage(), is("EOS is disabled [test]"));
}
@Test
public void shouldThrowStreamsExceptionOnSendError() {
- mockProducer.sendException = new KafkaException("KABOOM!");
+ nonEosMockProducer.sendException = new KafkaException("KABOOM!");
final StreamsException thrown = assertThrows(
StreamsException.class,
- () -> aloStreamsProducer.send(record, null)
+ () -> nonEosStreamsProducer.send(record, null)
);
- assertThat(thrown.getCause(), is(mockProducer.sendException));
- assertThat(thrown.getMessage(), is("Error encountered sending record to topic topic [test, alo]"));
- assertThat(thrown.getCause(), is(mockProducer.sendException));
+ assertThat(thrown.getCause(), is(nonEosMockProducer.sendException));
+ assertThat(thrown.getMessage(), is("Error encountered trying to send record to topic topic [test]"));
}
@Test
public void shouldFailOnSendFatal() {
- mockProducer.sendException = new RuntimeException("KABOOM!");
+ nonEosMockProducer.sendException = new RuntimeException("KABOOM!");
final RuntimeException thrown = assertThrows(
RuntimeException.class,
- () -> aloStreamsProducer.send(record, null)
+ () -> nonEosStreamsProducer.send(record, null)
);
assertThat(thrown.getMessage(), is("KABOOM!"));
@@ -202,22 +224,23 @@ public void shouldFailOnSendFatal() {
public void shouldFailOnCommitIfEosDisabled() {
final IllegalStateException thrown = assertThrows(
IllegalStateException.class,
- () -> aloStreamsProducer.commitTransaction(null)
+ () -> nonEosStreamsProducer.commitTransaction(null)
);
- assertThat(thrown.getMessage(), is("EOS is disabled [test, alo]"));
+ assertThat(thrown.getMessage(), is("EOS is disabled [test]"));
}
@Test
public void shouldFailOnAbortIfEosDisabled() {
final IllegalStateException thrown = assertThrows(
IllegalStateException.class,
- aloStreamsProducer::abortTransaction
+ nonEosStreamsProducer::abortTransaction
);
- assertThat(thrown.getMessage(), is("EOS is disabled [test, alo]"));
+ assertThat(thrown.getMessage(), is("EOS is disabled [test]"));
}
+
// EOS tests
// functional tests
@@ -262,7 +285,7 @@ public void shouldBeginTxOnEosCommit() {
replay(producer);
final StreamsProducer streamsProducer =
- new StreamsProducer(producer, true, logContext, "appId");
+ new StreamsProducer(producer, true, "appId", logContext);
streamsProducer.initTransaction();
streamsProducer.commitTransaction(offsetsAndMetadata);
@@ -318,7 +341,7 @@ public void shouldSkipAbortTxOnEosAbortIfNotTxInFlight() {
replay(producer);
final StreamsProducer streamsProducer =
- new StreamsProducer(producer, true, logContext, "appId");
+ new StreamsProducer(producer, true, "appId", logContext);
streamsProducer.initTransaction();
streamsProducer.abortTransaction();
@@ -328,12 +351,22 @@ public void shouldSkipAbortTxOnEosAbortIfNotTxInFlight() {
// error handling tests
+ @Test
+ public void shouldFailIfApplicationIdIsNullOnEos() {
+ final IllegalArgumentException thrown = assertThrows(
+ IllegalArgumentException.class,
+ () -> new StreamsProducer(eosMockProducer, true, null, logContext)
+ );
+
+ assertThat(thrown.getMessage(), is("applicationId cannot be null if EOS is enabled"));
+ }
+
@Test
public void shouldThrowTimeoutExceptionOnEosInitTxTimeout() {
// use `mockProducer` instead of `eosMockProducer` to avoid double Tx-Init
- mockProducer.initTransactionException = new TimeoutException("KABOOM!");
+ nonEosMockProducer.initTransactionException = new TimeoutException("KABOOM!");
final StreamsProducer streamsProducer =
- new StreamsProducer(mockProducer, true, logContext, "appId");
+ new StreamsProducer(nonEosMockProducer, true, "appId", logContext);
final TimeoutException thrown = assertThrows(
TimeoutException.class,
@@ -346,25 +379,25 @@ public void shouldThrowTimeoutExceptionOnEosInitTxTimeout() {
@Test
public void shouldThrowStreamsExceptionOnEosInitError() {
// use `mockProducer` instead of `eosMockProducer` to avoid double Tx-Init
- mockProducer.initTransactionException = new KafkaException("KABOOM!");
+ nonEosMockProducer.initTransactionException = new KafkaException("KABOOM!");
final StreamsProducer streamsProducer =
- new StreamsProducer(mockProducer, true, logContext, "appId");
+ new StreamsProducer(nonEosMockProducer, true, "appId", logContext);
final StreamsException thrown = assertThrows(
StreamsException.class,
streamsProducer::initTransaction
);
- assertThat(thrown.getCause(), is(mockProducer.initTransactionException));
- assertThat(thrown.getMessage(), is("Error encountered while initializing transactions [test, eos]"));
+ assertThat(thrown.getCause(), is(nonEosMockProducer.initTransactionException));
+ assertThat(thrown.getMessage(), is("Error encountered trying to initialize transactions [test]"));
}
@Test
public void shouldFailOnEosInitFatal() {
// use `mockProducer` instead of `eosMockProducer` to avoid double Tx-Init
- mockProducer.initTransactionException = new RuntimeException("KABOOM!");
+ nonEosMockProducer.initTransactionException = new RuntimeException("KABOOM!");
final StreamsProducer streamsProducer =
- new StreamsProducer(mockProducer, true, logContext, "appId");
+ new StreamsProducer(nonEosMockProducer, true, "appId", logContext);
final RuntimeException thrown = assertThrows(
RuntimeException.class,
@@ -385,7 +418,7 @@ public void shouldThrowTaskMigrateExceptionOnEosBeginTxnFenced() {
assertThat(
thrown.getMessage(),
- is("Producer get fenced trying to begin a new transaction [test, eos];" +
+ is("Producer got fenced trying to begin a new transaction [test];" +
" it means all tasks belonging to this thread should be migrated.")
);
}
@@ -402,7 +435,7 @@ public void shouldThrowTaskMigrateExceptionOnEosBeginTxnError() {
assertThat(thrown.getCause(), is(eosMockProducer.beginTransactionException));
assertThat(
thrown.getMessage(),
- is("Producer encounter unexpected error trying to begin a new transaction [test, eos]")
+ is("Error encountered trying to begin a new transaction [test]")
);
}
@@ -433,7 +466,7 @@ public void shouldThrowTaskMigratedExceptionOnEosSendFenced() {
assertThat(thrown.getCause(), is(exception));
assertThat(
thrown.getMessage(),
- is("Producer cannot send records anymore since it got fenced [test, eos];" +
+ is("Producer got fenced trying to send a record [test];" +
" it means all tasks belonging to this thread should be migrated.")
);
}
@@ -452,7 +485,7 @@ public void shouldThrowTaskMigratedExceptionOnEosSendUnknownPid() {
assertThat(thrown.getCause(), is(exception));
assertThat(
thrown.getMessage(),
- is("Producer cannot send records anymore since it got fenced [test, eos];" +
+ is("Producer got fenced trying to send a record [test];" +
" it means all tasks belonging to this thread should be migrated.")
);
}
@@ -472,7 +505,7 @@ public void shouldThrowTaskMigrateExceptionOnEosSendOffsetFenced() {
assertThat(thrown.getCause(), is(eosMockProducer.sendOffsetsToTransactionException));
assertThat(
thrown.getMessage(),
- is("Producer get fenced trying to commit a transaction [test, eos];" +
+ is("Producer got fenced trying to commit a transaction [test];" +
" it means all tasks belonging to this thread should be migrated.")
);
}
@@ -491,7 +524,7 @@ public void shouldThrowStreamsExceptionOnEosSendOffsetError() {
assertThat(thrown.getCause(), is(eosMockProducer.sendOffsetsToTransactionException));
assertThat(
thrown.getMessage(),
- is("Producer encounter unexpected error trying to commit a transaction [test, eos]")
+ is("Error encountered trying to commit a transaction [test]")
);
}
@@ -523,7 +556,7 @@ public void shouldThrowTaskMigrateExceptionOnEosCommitTxFenced() {
assertThat(thrown.getCause(), is(eosMockProducer.commitTransactionException));
assertThat(
thrown.getMessage(),
- is("Producer get fenced trying to commit a transaction [test, eos];" +
+ is("Producer got fenced trying to commit a transaction [test];" +
" it means all tasks belonging to this thread should be migrated.")
);
}
@@ -540,7 +573,7 @@ public void shouldThrowStreamsExceptionOnEosCommitTxTimeout() {
assertThat(eosMockProducer.sentOffsets(), is(true));
assertThat(thrown.getCause(), is(eosMockProducer.commitTransactionException));
- assertThat(thrown.getMessage(), is("Timed out while committing a transaction [test, eos]"));
+ assertThat(thrown.getMessage(), is("Timed out trying to commit a transaction [test]"));
}
@Test
@@ -556,7 +589,7 @@ public void shouldThrowStreamsExceptionOnEosCommitTxError() {
assertThat(thrown.getCause(), is(eosMockProducer.commitTransactionException));
assertThat(
thrown.getMessage(),
- is("Producer encounter unexpected error trying to commit a transaction [test, eos]")
+ is("Error encountered trying to commit a transaction [test]")
);
}
@@ -585,7 +618,7 @@ public void shouldSwallowExceptionOnEosAbortTxFenced() {
replay(producer);
final StreamsProducer streamsProducer =
- new StreamsProducer(producer, true, logContext, "appId");
+ new StreamsProducer(producer, true, "appId", logContext);
streamsProducer.initTransaction();
// call `send()` to start a transaction
streamsProducer.send(record, null);
@@ -606,7 +639,7 @@ public void shouldThrowStreamsExceptionOnEosAbortTxError() {
assertThat(thrown.getCause(), is(eosMockProducer.abortTransactionException));
assertThat(
thrown.getMessage(),
- is("Producer encounter unexpected error trying to abort a transaction [test, eos]")
+ is("Error encounter trying to abort a transaction [test]")
);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
index 2f7681d406b9d..d8bca5e6f35cf 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
@@ -17,13 +17,16 @@
package org.apache.kafka.streams.processor.internals;
import java.util.HashSet;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.DeleteRecordsResult;
import org.apache.kafka.clients.admin.DeletedRecords;
import org.apache.kafka.clients.admin.RecordsToDelete;
+import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
@@ -36,6 +39,7 @@
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.LockException;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
@@ -50,12 +54,14 @@
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
+import org.junit.function.ThrowingRunnable;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
+import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
import java.util.LinkedList;
@@ -79,12 +85,14 @@
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.easymock.EasyMock.resetToStrict;
import static org.easymock.EasyMock.verify;
import static org.hamcrest.CoreMatchers.hasItem;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.core.IsEqual.equalTo;
@@ -110,6 +118,10 @@ public class TaskManagerTest {
private final TopicPartition t1p2 = new TopicPartition(topic1, 2);
private final Set taskId02Partitions = mkSet(t1p2);
+ private final TaskId taskId03 = new TaskId(0, 3);
+ private final TopicPartition t1p3 = new TopicPartition(topic1, 3);
+ private final Set taskId03Partitions = mkSet(t1p3);
+
private final TaskId taskId10 = new TaskId(1, 0);
@Mock(type = MockType.STRICT)
@@ -144,7 +156,8 @@ public void setUp() {
standbyTaskCreator,
topologyBuilder,
adminClient,
- stateDirectory
+ stateDirectory,
+ false
);
taskManager.setMainConsumer(consumer);
}
@@ -155,10 +168,8 @@ public void shouldIdempotentlyUpdateSubscriptionFromActiveAssignment() {
final Map> assignment = mkMap(mkEntry(taskId01, mkSet(t1p1, newTopicPartition)));
expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andReturn(emptyList()).anyTimes();
-
topologyBuilder.addSubscribedTopicsFromAssignment(eq(asList(t1p1, newTopicPartition)), anyString());
expectLastCall();
-
replay(activeTaskCreator, topologyBuilder);
taskManager.handleAssignment(assignment, emptyMap());
@@ -341,26 +352,37 @@ public void shouldPinOffsetSumToLongMaxValueInCaseOfOverflow() throws IOExceptio
@Test
public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
- final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final Map offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ task00.setCommittableOffsetsAndMetadata(offsets);
+ // first `handleAssignment`
expectRestoreToBeCompleted(consumer, changeLogReader);
expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00)).anyTimes();
expect(activeTaskCreator.createTasks(anyObject(), eq(emptyMap()))).andReturn(emptyList()).anyTimes();
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
expectLastCall();
expect(standbyTaskCreator.createTasks(anyObject())).andReturn(emptyList()).anyTimes();
-
topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString());
expectLastCall().anyTimes();
+ // `handleRevocation`
+ consumer.commitSync(offsets);
+ expectLastCall();
+
+ // second `handleAssignment`
+ consumer.commitSync(offsets);
+ expectLastCall();
+
replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader);
taskManager.handleAssignment(taskId00Assignment, emptyMap());
-
assertThat(taskManager.tryToCompleteRestoration(), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
+
taskManager.handleRevocation(taskId00Partitions);
assertThat(task00.state(), is(Task.State.SUSPENDED));
+
taskManager.handleAssignment(emptyMap(), emptyMap());
assertThat(task00.state(), is(Task.State.CLOSED));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
@@ -368,27 +390,61 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
}
@Test
- public void shouldCloseActiveTasksWhenHandlingLostTasks() {
- final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
- final Task task01 = new StateMachineTask(taskId01, taskId01Partitions, false);
+ public void shouldCloseDirtyActiveUnassignedSuspendedTasksWhenErrorCommittingRevokedTask() {
+ final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true) {
+ @Override
+ public Map committableOffsetsAndMetadata() {
+ throw new RuntimeException("KABOOM!");
+ }
+ };
+ // first `handleAssignment`
expectRestoreToBeCompleted(consumer, changeLogReader);
expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00)).anyTimes();
expect(activeTaskCreator.createTasks(anyObject(), eq(emptyMap()))).andReturn(emptyList()).anyTimes();
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
expectLastCall();
- expect(standbyTaskCreator.createTasks(eq(taskId01Assignment))).andReturn(singletonList(task01)).anyTimes();
+ expect(standbyTaskCreator.createTasks(anyObject())).andReturn(emptyList()).anyTimes();
+ topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString());
+ expectLastCall().anyTimes();
+
+ replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader);
+
+ taskManager.handleAssignment(taskId00Assignment, emptyMap());
+
+ final RuntimeException thrown = assertThrows(
+ RuntimeException.class,
+ () -> taskManager.handleAssignment(emptyMap(), emptyMap())
+ );
+
+ assertThat(task00.state(), is(Task.State.CLOSED));
+ assertThat(thrown.getMessage(), is("Unexpected failure to close 1 task(s) [[0_0]]. First unexpected exception (for task 0_0) follows."));
+ assertThat(thrown.getCause().getMessage(), is("KABOOM!"));
+ }
+ @Test
+ public void shouldCloseActiveTasksWhenHandlingLostTasks() {
+ final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final Task task01 = new StateMachineTask(taskId01, taskId01Partitions, false);
+
+ // `handleAssignment`
+ expectRestoreToBeCompleted(consumer, changeLogReader);
+ expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00)).anyTimes();
+ expect(standbyTaskCreator.createTasks(eq(taskId01Assignment))).andReturn(singletonList(task01)).anyTimes();
topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString());
expectLastCall().anyTimes();
+ // `handleLostAll`
+ activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
+ expectLastCall();
+
replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader);
taskManager.handleAssignment(taskId00Assignment, taskId01Assignment);
-
assertThat(taskManager.tryToCompleteRestoration(), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
assertThat(task01.state(), is(Task.State.RUNNING));
+
taskManager.handleLostAll();
assertThat(task00.state(), is(Task.State.CLOSED));
assertThat(task01.state(), is(Task.State.RUNNING));
@@ -396,33 +452,66 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() {
assertThat(taskManager.standbyTaskMap(), is(singletonMap(taskId01, task01)));
}
+ @Test
+ public void shouldThrowWhenHandlingClosingTasksOnProducerCloseError() {
+ final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final Map offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ task00.setCommittableOffsetsAndMetadata(offsets);
+
+ // `handleAssignment`
+ expectRestoreToBeCompleted(consumer, changeLogReader);
+ expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00)).anyTimes();
+ expect(standbyTaskCreator.createTasks(anyObject())).andReturn(emptyList()).anyTimes();
+ topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString());
+ expectLastCall().anyTimes();
+
+ // `handleAssignment`
+ consumer.commitSync(offsets);
+ expectLastCall();
+ activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
+ expectLastCall().andThrow(new RuntimeException("KABOOM!"));
+
+ replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader);
+
+ taskManager.handleAssignment(taskId00Assignment, emptyMap());
+ assertThat(taskManager.tryToCompleteRestoration(), is(true));
+ assertThat(task00.state(), is(Task.State.RUNNING));
+
+ final RuntimeException thrown = assertThrows(
+ RuntimeException.class,
+ () -> taskManager.handleAssignment(emptyMap(), emptyMap())
+ );
+
+ assertThat(thrown.getMessage(), is("Unexpected failure to close 1 task(s) [[0_0]]. First unexpected exception (for task 0_0) follows."));
+ assertThat(thrown.getCause(), instanceOf(RuntimeException.class));
+ assertThat(thrown.getCause().getMessage(), is("KABOOM!"));
+ }
+
@Test
public void shouldReviveCorruptTasks() {
final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class);
stateManager.markChangelogAsCorrupted(taskId00Partitions);
replay(stateManager);
+
final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager);
+ // `handleAssignment`
expectRestoreToBeCompleted(consumer, changeLogReader);
expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00)).anyTimes();
- expect(activeTaskCreator.createTasks(anyObject(), eq(emptyMap()))).andReturn(emptyList()).anyTimes();
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
- expectLastCall();
- expect(standbyTaskCreator.createTasks(anyObject())).andReturn(emptyList()).anyTimes();
-
topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString());
expectLastCall().anyTimes();
- replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader);
+ replay(activeTaskCreator, topologyBuilder, consumer, changeLogReader);
taskManager.handleAssignment(taskId00Assignment, emptyMap());
-
assertThat(taskManager.tryToCompleteRestoration(), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
+
taskManager.handleCorruption(singletonMap(taskId00, taskId00Partitions));
assertThat(task00.state(), is(Task.State.CREATED));
assertThat(taskManager.activeTaskMap(), is(singletonMap(taskId00, task00)));
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
+
verify(stateManager);
}
@@ -431,33 +520,30 @@ public void shouldReviveCorruptTasksEvenIfTheyCannotCloseClean() {
final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class);
stateManager.markChangelogAsCorrupted(taskId00Partitions);
replay(stateManager);
+
final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new RuntimeException("oops");
}
};
expectRestoreToBeCompleted(consumer, changeLogReader);
expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00)).anyTimes();
- expect(activeTaskCreator.createTasks(anyObject(), eq(emptyMap()))).andReturn(emptyList()).anyTimes();
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
- expectLastCall();
- expect(standbyTaskCreator.createTasks(anyObject())).andReturn(emptyList()).anyTimes();
-
topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(), anyString());
expectLastCall().anyTimes();
- replay(activeTaskCreator, standbyTaskCreator, topologyBuilder, consumer, changeLogReader);
+ replay(activeTaskCreator, topologyBuilder, consumer, changeLogReader);
taskManager.handleAssignment(taskId00Assignment, emptyMap());
-
assertThat(taskManager.tryToCompleteRestoration(), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
+
taskManager.handleCorruption(singletonMap(taskId00, taskId00Partitions));
assertThat(task00.state(), is(Task.State.CREATED));
assertThat(taskManager.activeTaskMap(), is(singletonMap(taskId00, task00)));
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
+
verify(stateManager);
}
@@ -467,10 +553,14 @@ public void shouldCloseStandbyUnassignedTasksWhenCreatingNewTasks() {
expectRestoreToBeCompleted(consumer, changeLogReader);
expect(standbyTaskCreator.createTasks(eq(taskId00Assignment))).andReturn(singletonList(task00)).anyTimes();
+ consumer.commitSync(Collections.emptyMap());
+ expectLastCall();
replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader);
+
taskManager.handleAssignment(emptyMap(), taskId00Assignment);
assertThat(taskManager.tryToCompleteRestoration(), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
+
taskManager.handleAssignment(emptyMap(), emptyMap());
assertThat(task00.state(), is(Task.State.CLOSED));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
@@ -548,6 +638,8 @@ public void initializeIfNeeded() {
}
};
+ consumer.commitSync(Collections.emptyMap());
+ expectLastCall();
expect(changeLogReader.completedChangelogs()).andReturn(emptySet());
expect(consumer.assignment()).andReturn(emptySet());
consumer.resume(eq(emptySet()));
@@ -587,6 +679,8 @@ public void completeRestoration() {
}
};
+ consumer.commitSync(Collections.emptyMap());
+ expectLastCall();
expect(changeLogReader.completedChangelogs()).andReturn(emptySet());
expect(consumer.assignment()).andReturn(emptySet());
consumer.resume(eq(emptySet()));
@@ -614,12 +708,17 @@ public void completeRestoration() {
@Test
public void shouldSuspendActiveTasks() {
- final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final Map offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ task00.setCommittableOffsetsAndMetadata(offsets);
expectRestoreToBeCompleted(consumer, changeLogReader);
expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00));
+ consumer.commitSync(offsets);
+ expectLastCall();
replay(activeTaskCreator, consumer, changeLogReader);
+
taskManager.handleAssignment(taskId00Assignment, emptyMap());
assertThat(taskManager.tryToCompleteRestoration(), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
@@ -629,10 +728,24 @@ public void shouldSuspendActiveTasks() {
}
@Test
- public void shouldPassUpIfExceptionDuringSuspend() {
+ public void shouldNotCommitCreatedTasksOnSuspend() {
+ final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+
+ expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00));
+ replay(activeTaskCreator, consumer, changeLogReader);
+
+ taskManager.handleAssignment(taskId00Assignment, emptyMap());
+ assertThat(task00.state(), is(Task.State.CREATED));
+
+ taskManager.handleRevocation(taskId00Partitions);
+ assertThat(task00.state(), is(Task.State.CREATED));
+ }
+
+ @Test
+ public void shouldPassUpIfExceptionDuringPrepareSuspend() {
final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true) {
@Override
- public void suspend() {
+ public void prepareSuspend() {
throw new RuntimeException("KABOOM!");
}
};
@@ -647,6 +760,8 @@ public void suspend() {
assertThrows(RuntimeException.class, () -> taskManager.handleRevocation(taskId00Partitions));
assertThat(task00.state(), is(Task.State.RUNNING));
+
+ verify(consumer);
}
@Test
@@ -655,7 +770,8 @@ public void shouldCloseActiveTasksAndPropagateExceptionsOnCleanShutdown() {
final Map> assignment = mkMap(
mkEntry(taskId00, taskId00Partitions),
mkEntry(taskId01, taskId01Partitions),
- mkEntry(taskId02, taskId02Partitions)
+ mkEntry(taskId02, taskId02Partitions),
+ mkEntry(taskId03, taskId03Partitions)
);
final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true) {
@Override
@@ -663,17 +779,65 @@ public Collection changelogPartitions() {
return singletonList(changelog);
}
};
+ final AtomicBoolean prepareClosedDirtyTask01 = new AtomicBoolean(false);
+ final AtomicBoolean prepareClosedDirtyTask02 = new AtomicBoolean(false);
+ final AtomicBoolean prepareClosedDirtyTask03 = new AtomicBoolean(false);
+ final AtomicBoolean closedDirtyTask01 = new AtomicBoolean(false);
+ final AtomicBoolean closedDirtyTask02 = new AtomicBoolean(false);
+ final AtomicBoolean closedDirtyTask03 = new AtomicBoolean(false);
final Task task01 = new StateMachineTask(taskId01, taskId01Partitions, true) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new TaskMigratedException("migrated", new RuntimeException("cause"));
}
+
+ @Override
+ public void prepareCloseDirty() {
+ super.prepareCloseDirty();
+ prepareClosedDirtyTask01.set(true);
+ }
+
+ @Override
+ public void closeDirty() {
+ super.closeDirty();
+ closedDirtyTask01.set(true);
+ }
};
final Task task02 = new StateMachineTask(taskId02, taskId02Partitions, true) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new RuntimeException("oops");
}
+
+ @Override
+ public void prepareCloseDirty() {
+ super.prepareCloseDirty();
+ prepareClosedDirtyTask02.set(true);
+ }
+
+ @Override
+ public void closeDirty() {
+ super.closeDirty();
+ closedDirtyTask02.set(true);
+ }
+ };
+ final Task task03 = new StateMachineTask(taskId03, taskId03Partitions, true) {
+ @Override
+ public Map committableOffsetsAndMetadata() {
+ throw new RuntimeException("oops");
+ }
+
+ @Override
+ public void prepareCloseDirty() {
+ super.prepareCloseDirty();
+ prepareClosedDirtyTask03.set(true);
+ }
+
+ @Override
+ public void closeDirty() {
+ super.closeDirty();
+ closedDirtyTask03.set(true);
+ }
};
resetToStrict(changeLogReader);
@@ -683,13 +847,16 @@ public void closeClean() {
// make sure we also remove the changelog partitions from the changelog reader
changeLogReader.remove(eq(singletonList(changelog)));
expectLastCall();
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andReturn(asList(task00, task01, task02)).anyTimes();
+ expect(activeTaskCreator.createTasks(anyObject(), eq(assignment)))
+ .andReturn(asList(task00, task01, task02, task03)).anyTimes();
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId00));
expectLastCall();
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId01));
expectLastCall();
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId02));
expectLastCall();
+ activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId03));
+ expectLastCall();
activeTaskCreator.closeThreadProducerIfNeeded();
expectLastCall();
expect(standbyTaskCreator.createTasks(eq(emptyMap()))).andReturn(emptyList()).anyTimes();
@@ -700,29 +867,42 @@ public void closeClean() {
assertThat(task00.state(), is(Task.State.CREATED));
assertThat(task01.state(), is(Task.State.CREATED));
assertThat(task02.state(), is(Task.State.CREATED));
+ assertThat(task03.state(), is(Task.State.CREATED));
taskManager.tryToCompleteRestoration();
assertThat(task00.state(), is(Task.State.RESTORING));
assertThat(task01.state(), is(Task.State.RUNNING));
assertThat(task02.state(), is(Task.State.RUNNING));
+ assertThat(task03.state(), is(Task.State.RUNNING));
assertThat(
taskManager.activeTaskMap(),
Matchers.equalTo(
mkMap(
mkEntry(taskId00, task00),
mkEntry(taskId01, task01),
- mkEntry(taskId02, task02)
+ mkEntry(taskId02, task02),
+ mkEntry(taskId03, task03)
)
)
);
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
- final RuntimeException exception = assertThrows(RuntimeException.class, () -> taskManager.shutdown(true));
+ final RuntimeException exception = assertThrows(
+ RuntimeException.class,
+ () -> taskManager.shutdown(true)
+ );
+ assertThat(prepareClosedDirtyTask01.get(), is(true));
+ assertThat(closedDirtyTask01.get(), is(true));
+ assertThat(prepareClosedDirtyTask02.get(), is(true));
+ assertThat(closedDirtyTask02.get(), is(true));
+ assertThat(prepareClosedDirtyTask03.get(), is(true));
+ assertThat(closedDirtyTask03.get(), is(true));
assertThat(task00.state(), is(Task.State.CLOSED));
assertThat(task01.state(), is(Task.State.CLOSED));
assertThat(task02.state(), is(Task.State.CLOSED));
+ assertThat(task03.state(), is(Task.State.CLOSED));
assertThat(exception.getMessage(), is("Unexpected exception while closing task"));
assertThat(exception.getCause().getMessage(), is("oops"));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
@@ -737,12 +917,14 @@ public void shouldCloseActiveTasksAndPropagateTaskProducerExceptionsOnCleanShutd
final Map> assignment = mkMap(
mkEntry(taskId00, taskId00Partitions)
);
- final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, true) {
+ final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true) {
@Override
public Collection changelogPartitions() {
return singletonList(changelog);
}
};
+ final Map offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ task00.setCommittableOffsetsAndMetadata(offsets);
resetToStrict(changeLogReader);
changeLogReader.transitToRestoreActive();
@@ -859,13 +1041,13 @@ public Collection changelogPartitions() {
};
final Task task01 = new StateMachineTask(taskId01, taskId01Partitions, true) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new TaskMigratedException("migrated", new RuntimeException("cause"));
}
};
final Task task02 = new StateMachineTask(taskId02, taskId02Partitions, true) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new RuntimeException("oops");
}
};
@@ -928,28 +1110,32 @@ public void shouldCloseStandbyTasksOnShutdown() {
final Map> assignment = singletonMap(taskId00, taskId00Partitions);
final Task task00 = new StateMachineTask(taskId00, taskId00Partitions, false);
+ // `handleAssignment`
+ expect(standbyTaskCreator.createTasks(eq(assignment))).andReturn(singletonList(task00)).anyTimes();
+
+ // `tryToCompleteRestoration`
expect(changeLogReader.completedChangelogs()).andReturn(emptySet());
expect(consumer.assignment()).andReturn(emptySet());
consumer.resume(eq(emptySet()));
expectLastCall();
- expect(activeTaskCreator.createTasks(anyObject(), eq(emptyMap()))).andReturn(emptyList()).anyTimes();
+
+ // `shutdown`
+ consumer.commitSync(Collections.emptyMap());
+ expectLastCall();
activeTaskCreator.closeThreadProducerIfNeeded();
expectLastCall();
- expect(standbyTaskCreator.createTasks(eq(assignment))).andReturn(singletonList(task00)).anyTimes();
+
replay(consumer, activeTaskCreator, standbyTaskCreator, changeLogReader);
taskManager.handleAssignment(emptyMap(), assignment);
-
assertThat(task00.state(), is(Task.State.CREATED));
taskManager.tryToCompleteRestoration();
-
assertThat(task00.state(), is(Task.State.RUNNING));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.equalTo(singletonMap(taskId00, task00)));
taskManager.shutdown(true);
-
assertThat(task00.state(), is(Task.State.CLOSED));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
@@ -1011,6 +1197,8 @@ public void shouldHandleRebalanceEvents() {
@Test
public void shouldCommitActiveAndStandbyTasks() {
final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final Map offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ task00.setCommittableOffsetsAndMetadata(offsets);
final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, false);
expectRestoreToBeCompleted(consumer, changeLogReader);
@@ -1018,6 +1206,8 @@ public void shouldCommitActiveAndStandbyTasks() {
.andReturn(singletonList(task00)).anyTimes();
expect(standbyTaskCreator.createTasks(eq(taskId01Assignment)))
.andReturn(singletonList(task01)).anyTimes();
+ consumer.commitSync(offsets);
+ expectLastCall();
replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader);
@@ -1031,6 +1221,8 @@ public void shouldCommitActiveAndStandbyTasks() {
task01.setCommitNeeded();
assertThat(taskManager.commitAll(), equalTo(2));
+ assertThat(task00.commitNeeded, is(false));
+ assertThat(task01.commitNeeded, is(false));
}
@Test
@@ -1070,11 +1262,62 @@ public void shouldNotCommitActiveAndStandbyTasksWhileRebalanceInProgress() throw
);
}
+ @Test
+ public void shouldCommitViaConsumerIfEosDisabled() {
+ final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, true);
+ final Map offsets = singletonMap(t1p1, new OffsetAndMetadata(0L, null));
+ task01.setCommittableOffsetsAndMetadata(offsets);
+ task01.setCommitNeeded();
+ taskManager.tasks().put(taskId01, task01);
+
+ consumer.commitSync(offsets);
+ expectLastCall();
+ replay(consumer);
+
+ taskManager.commitAll();
+
+ verify(consumer);
+ }
+
+ @Test
+ public void shouldCommitViaProducerIfEosEnabled() {
+ final StreamsProducer producer = mock(StreamsProducer.class);
+ final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(new Metrics(), "clientId", StreamsConfig.METRICS_LATEST);
+ taskManager = new TaskManager(
+ changeLogReader,
+ UUID.randomUUID(),
+ "taskManagerTest",
+ streamsMetrics,
+ activeTaskCreator,
+ standbyTaskCreator,
+ topologyBuilder,
+ adminClient,
+ stateDirectory,
+ true
+ );
+ taskManager.setMainConsumer(consumer);
+
+ final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, true);
+ final Map offsets = singletonMap(t1p1, new OffsetAndMetadata(0L, null));
+ task01.setCommittableOffsetsAndMetadata(offsets);
+ task01.setCommitNeeded();
+ taskManager.tasks().put(taskId01, task01);
+
+ expect(activeTaskCreator.streamsProducerForTask(taskId01)).andReturn(producer);
+ producer.commitTransaction(offsets);
+ expectLastCall();
+ replay(activeTaskCreator, producer);
+
+ taskManager.commitAll();
+
+ verify(producer);
+ }
+
@Test
public void shouldPropagateExceptionFromActiveCommit() {
final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true) {
@Override
- public void commit() {
+ public void prepareCommit() {
throw new RuntimeException("opsh.");
}
};
@@ -1101,7 +1344,7 @@ public void commit() {
public void shouldPropagateExceptionFromStandbyCommit() {
final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, false) {
@Override
- public void commit() {
+ public void prepareCommit() {
throw new RuntimeException("opsh.");
}
};
@@ -1231,28 +1474,46 @@ public void shouldIgnorePurgeDataErrors() {
@Test
public void shouldMaybeCommitActiveTasks() {
final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final Map offsets0 = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ task00.setCommittableOffsetsAndMetadata(offsets0);
final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, true);
+ final Map offsets1 = singletonMap(t1p1, new OffsetAndMetadata(1L, null));
+ task01.setCommittableOffsetsAndMetadata(offsets1);
final StateMachineTask task02 = new StateMachineTask(taskId02, taskId02Partitions, true);
+ final Map offsets2 = singletonMap(t1p2, new OffsetAndMetadata(2L, null));
+ task02.setCommittableOffsetsAndMetadata(offsets2);
+ final StateMachineTask task03 = new StateMachineTask(taskId03, taskId03Partitions, false);
+ final Map offsets3 = singletonMap(t1p3, new OffsetAndMetadata(3L, null));
+ task03.setCommittableOffsetsAndMetadata(offsets3);
- final Map> assignment = mkMap(
+ final Map> assignmentActive = mkMap(
mkEntry(taskId00, taskId00Partitions),
mkEntry(taskId01, taskId01Partitions),
mkEntry(taskId02, taskId02Partitions)
);
+ final Map> assignmentStandby = mkMap(
+ mkEntry(taskId03, taskId03Partitions)
+ );
+
expectRestoreToBeCompleted(consumer, changeLogReader);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment)))
+ expect(activeTaskCreator.createTasks(anyObject(), eq(assignmentActive)))
.andReturn(asList(task00, task01, task02)).anyTimes();
+ expect(standbyTaskCreator.createTasks(eq(assignmentStandby)))
+ .andReturn(singletonList(task03)).anyTimes();
+ consumer.commitSync(offsets0);
+ expectLastCall();
replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader);
- taskManager.handleAssignment(assignment, emptyMap());
+ taskManager.handleAssignment(assignmentActive, assignmentStandby);
assertThat(taskManager.tryToCompleteRestoration(), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
assertThat(task01.state(), is(Task.State.RUNNING));
assertThat(task02.state(), is(Task.State.RUNNING));
+ assertThat(task03.state(), is(Task.State.RUNNING));
task00.setCommitNeeded();
task00.setCommitRequested();
@@ -1261,6 +1522,9 @@ public void shouldMaybeCommitActiveTasks() {
task02.setCommitRequested();
+ task03.setCommitNeeded();
+ task03.setCommitRequested();
+
assertThat(taskManager.maybeCommitActiveTasksPerUserRequested(), equalTo(1));
}
@@ -1450,9 +1714,13 @@ public void shouldHaveRemainingPartitionsUncleared() {
final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
+ final Map offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ task00.setCommittableOffsetsAndMetadata(offsets);
expectRestoreToBeCompleted(consumer, changeLogReader);
expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andReturn(singletonList(task00));
+ consumer.commitSync(offsets);
+ expectLastCall();
replay(activeTaskCreator, consumer, changeLogReader);
taskManager.handleAssignment(taskId00Assignment, emptyMap());
@@ -1473,14 +1741,14 @@ public void shouldHaveRemainingPartitionsUncleared() {
public void shouldThrowTaskMigratedWhenAllTaskCloseExceptionsAreTaskMigrated() {
final StateMachineTask migratedTask01 = new StateMachineTask(taskId01, taskId01Partitions, false) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new TaskMigratedException("t1 close exception", new RuntimeException());
}
};
final StateMachineTask migratedTask02 = new StateMachineTask(taskId02, taskId02Partitions, false) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new TaskMigratedException("t2 close exception", new RuntimeException());
}
};
@@ -1500,14 +1768,14 @@ public void closeClean() {
public void shouldThrowRuntimeExceptionWhenEncounteredUnknownExceptionDuringTaskClose() {
final StateMachineTask migratedTask01 = new StateMachineTask(taskId01, taskId01Partitions, false) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new TaskMigratedException("t1 close exception", new RuntimeException());
}
};
final StateMachineTask migratedTask02 = new StateMachineTask(taskId02, taskId02Partitions, false) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new IllegalStateException("t2 illegal state exception", new RuntimeException());
}
};
@@ -1529,14 +1797,14 @@ public void closeClean() {
public void shouldThrowSameKafkaExceptionWhenEncounteredDuringTaskClose() {
final StateMachineTask migratedTask01 = new StateMachineTask(taskId01, taskId01Partitions, false) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new TaskMigratedException("t1 close exception", new RuntimeException());
}
};
final StateMachineTask migratedTask02 = new StateMachineTask(taskId02, taskId02Partitions, false) {
@Override
- public void closeClean() {
+ public Map prepareCloseClean() {
throw new KafkaException("Kaboom for t2!", new RuntimeException());
}
};
@@ -1645,8 +1913,125 @@ private static void expectConsumerAssignmentPaused(final Consumer taskManager.commitAll()
+ );
+
+ assertThat(thrown.getCause(), instanceOf(CommitFailedException.class));
+ assertThat(thrown.getMessage(), equalTo("Consumer committing offsets failed, indicating the corresponding thread is no longer part of the group; it means all tasks belonging to this thread should be migrated."));
+ assertThat(task01.state(), is(Task.State.CREATED));
+ }
+
+ @Test
+ public void shouldThrowStreamsExceptionOnCommitTimeout() {
+ final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, true);
+ task01.setCommitNeeded();
+ taskManager.tasks().put(taskId01, task01);
+
+ consumer.commitSync(Collections.emptyMap());
+ expectLastCall().andThrow(new TimeoutException());
+ replay(consumer);
+
+ final StreamsException thrown = assertThrows(
+ StreamsException.class,
+ () -> taskManager.commitAll()
+ );
+
+ assertThat(thrown.getCause(), instanceOf(TimeoutException.class));
+ assertThat(thrown.getMessage(), equalTo("Timed out while committing offsets via consumer"));
+ assertThat(task01.state(), is(Task.State.CREATED));
+ }
+
+ @Test
+ public void shouldStreamsExceptionOnCommitError() {
+ final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, true);
+ task01.setCommitNeeded();
+ taskManager.tasks().put(taskId01, task01);
+
+ consumer.commitSync(Collections.emptyMap());
+ expectLastCall().andThrow(new KafkaException());
+ replay(consumer);
+
+ final StreamsException thrown = assertThrows(
+ StreamsException.class,
+ () -> taskManager.commitAll()
+ );
+
+ assertThat(thrown.getCause(), instanceOf(KafkaException.class));
+ assertThat(thrown.getMessage(), equalTo("Error encountered committing offsets via consumer"));
+ assertThat(task01.state(), is(Task.State.CREATED));
+ }
+
+ @Test
+ public void shouldFailOnCommitFatal() {
+ final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, true);
+ task01.setCommitNeeded();
+ taskManager.tasks().put(taskId01, task01);
+
+ consumer.commitSync(Collections.emptyMap());
+ expectLastCall().andThrow(new RuntimeException("KABOOM"));
+ replay(consumer);
+
+ final RuntimeException thrown = assertThrows(
+ RuntimeException.class,
+ () -> taskManager.commitAll()
+ );
+
+ assertThat(thrown.getMessage(), equalTo("KABOOM"));
+ assertThat(task01.state(), is(Task.State.CREATED));
+ }
+
+ @Test
+ public void shouldNotCloseTasksIfCommittingFailsDuringAssignment() {
+ shouldNotCloseTaskIfCommitFailsDuringAction(() -> taskManager.handleAssignment(Collections.emptyMap(), Collections.emptyMap()));
+ }
+
+ @Test
+ public void shouldNotCloseTasksIfCommittingFailsDuringRevocation() {
+ shouldNotCloseTaskIfCommitFailsDuringAction(() -> taskManager.handleRevocation(singletonList(t1p0)));
+ }
+
+ @Test
+ public void shouldNotCloseTasksIfCommittingFailsDuringShutdown() {
+ shouldNotCloseTaskIfCommitFailsDuringAction(() -> taskManager.shutdown(true));
+ }
+
+ private void shouldNotCloseTaskIfCommitFailsDuringAction(final ThrowingRunnable action) {
+ final Map offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
+ final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true) {
+ @Override
+ public Map committableOffsetsAndMetadata() {
+ return offsets;
+ }
+ };
+
+ expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment)))
+ .andReturn(singletonList(task00));
+ consumer.commitSync(offsets);
+ expectLastCall().andThrow(new RuntimeException("KABOOM!"));
+ replay(activeTaskCreator, consumer);
+
+ taskManager.handleAssignment(taskId00Assignment, Collections.emptyMap());
+
+ final RuntimeException thrown = assertThrows(RuntimeException.class, action);
+
+ assertThat(thrown.getMessage(), is("KABOOM!"));
+ assertThat(task00.state(), is(Task.State.CREATED));
+ }
+
private static void expectRestoreToBeCompleted(final Consumer consumer,
- final ChangelogReader changeLogReader) {
+ final ChangelogReader changeLogReader) {
final Set assignment = singleton(new TopicPartition("assignment", 0));
expect(consumer.assignment()).andReturn(assignment);
consumer.resume(assignment);
@@ -1683,6 +2068,7 @@ private static class StateMachineTask extends AbstractTask implements Task {
private final boolean active;
private boolean commitNeeded = false;
private boolean commitRequested = false;
+ private Map committableOffsets = Collections.emptyMap();
private Map purgeableOffsets;
private Map changelogOffsets;
private Map>> queue = new HashMap<>();
@@ -1732,7 +2118,15 @@ public boolean commitRequested() {
}
@Override
- public void commit() {}
+ public void prepareCommit(){}
+
+ @Override
+ public void postCommit() {
+ commitNeeded = false;
+ }
+
+ @Override
+ public void prepareSuspend() {}
@Override
public void suspend() {
@@ -1747,17 +2141,32 @@ public void resume() {
}
@Override
- public void closeClean() {
- transitionTo(State.CLOSING);
+ public Map prepareCloseClean() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public void prepareCloseDirty() {}
+
+ @Override
+ public void closeClean(final Map checkpoint) {
transitionTo(State.CLOSED);
}
@Override
public void closeDirty() {
- transitionTo(State.CLOSING);
transitionTo(State.CLOSED);
}
+ void setCommittableOffsetsAndMetadata(final Map committableOffsets) {
+ this.committableOffsets = committableOffsets;
+ }
+
+ @Override
+ public Map committableOffsetsAndMetadata() {
+ return committableOffsets;
+ }
+
@Override
public StateStore getStore(final String name) {
return null;
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
index 1c8bf9829ace5..b212f3711c384 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -200,10 +200,8 @@ private KeyValueStoreTestDriver(final StateSerdes serdes) {
final RecordCollector recordCollector = new RecordCollectorImpl(
logContext,
new TaskId(0, 0),
- consumer,
- new StreamsProducer(producer, false, logContext, null),
+ new StreamsProducer(producer, false, null, logContext),
new DefaultProductionExceptionHandler(),
- false,
new MockStreamsMetrics(new Metrics())
) {
@Override
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
index 2d7db43b639bb..9b03fc0f28970 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
@@ -374,10 +374,13 @@ private StreamTask createStreamsTask(final StreamsConfig streamsConfig,
final RecordCollector recordCollector = new RecordCollectorImpl(
logContext,
taskId,
- clientSupplier.consumer,
- new StreamsProducer(clientSupplier.getProducer(new HashMap<>()), eosEnabled, logContext, streamsConfig.getString(StreamsConfig.APPLICATION_ID_CONFIG)),
+ new StreamsProducer(
+ clientSupplier.getProducer(new HashMap<>()),
+ eosEnabled,
+ streamsConfig.getString(StreamsConfig.APPLICATION_ID_CONFIG),
+ logContext
+ ),
streamsConfig.defaultProductionExceptionHandler(),
- eosEnabled,
new MockStreamsMetrics(metrics));
return new StreamTask(
taskId,
diff --git a/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java b/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java
index e34fab12657ad..2a781f034b261 100644
--- a/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java
+++ b/streams/src/test/java/org/apache/kafka/test/MockRecordCollector.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.test;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Headers;
@@ -36,9 +35,6 @@ public class MockRecordCollector implements RecordCollector {
// remember all records that are collected so far
private final List> collected = new LinkedList<>();
- // remember all commits that are submitted so far
- private final List> committed = new LinkedList<>();
-
// remember if flushed is called
private boolean flushed = false;
@@ -79,11 +75,6 @@ public void send(final String topic,
@Override
public void initialize() {}
- @Override
- public void commit(final Map offsets) {
- committed.add(offsets);
- }
-
@Override
public void flush() {
flushed = true;
@@ -101,10 +92,6 @@ public List> collected() {
return unmodifiableList(collected);
}
- public List> committed() {
- return unmodifiableList(committed);
- }
-
public boolean flushed() {
return flushed;
}
diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
index 65eb17a4bebab..d0ee43a39a61f 100644
--- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
+++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
@@ -19,6 +19,7 @@
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
@@ -65,7 +66,7 @@
import org.apache.kafka.streams.processor.internals.StoreChangelogReader;
import org.apache.kafka.streams.processor.internals.StreamTask;
import org.apache.kafka.streams.processor.internals.Task;
-import org.apache.kafka.streams.processor.internals.StreamsProducer;
+import org.apache.kafka.streams.processor.internals.TestDriverProducer;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.processor.internals.metrics.TaskMetrics;
import org.apache.kafka.streams.state.KeyValueStore;
@@ -213,7 +214,9 @@ public class TopologyTestDriver implements Closeable {
ProcessorTopology processorTopology;
ProcessorTopology globalTopology;
+ private final MockConsumer consumer;
private final MockProducer producer;
+ private final TestDriverProducer testDriverProducer;
private final Map partitionsByInputTopic = new HashMap<>();
private final Map globalPartitionsByInputTopic = new HashMap<>();
@@ -300,8 +303,10 @@ private TopologyTestDriver(final InternalTopologyBuilder builder,
final ThreadCache cache = new ThreadCache(
logContext,
Math.max(0, streamsConfig.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG)),
- streamsMetrics);
+ streamsMetrics
+ );
+ consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
final Serializer bytesSerializer = new ByteArraySerializer();
producer = new MockProducer(true, bytesSerializer, bytesSerializer) {
@Override
@@ -309,6 +314,12 @@ public List partitionsFor(final String topic) {
return Collections.singletonList(new PartitionInfo(topic, PARTITION_ID, null, null, null));
}
};
+ testDriverProducer = new TestDriverProducer(
+ producer,
+ eosEnabled,
+ streamsConfig.getString(StreamsConfig.APPLICATION_ID_CONFIG),
+ logContext
+ );
setupGlobalTask(streamsConfig, streamsMetrics, cache);
setupTask(streamsConfig, streamsMetrics, cache);
@@ -388,7 +399,8 @@ private void setupGlobalTask(final StreamsConfig streamsConfig,
globalConsumer,
stateDirectory,
stateRestoreListener,
- streamsConfig);
+ streamsConfig
+ );
final GlobalProcessorContextImpl globalProcessorContext =
new GlobalProcessorContextImpl(streamsConfig, globalStateManager, streamsMetrics, cache);
@@ -407,7 +419,8 @@ private void setupGlobalTask(final StreamsConfig streamsConfig,
-1L,
-1,
ProcessorContextImpl.NONEXIST_TOPIC,
- new RecordHeaders()));
+ new RecordHeaders())
+ );
} else {
globalStateManager = null;
globalStateTask = null;
@@ -418,7 +431,6 @@ private void setupTask(final StreamsConfig streamsConfig,
final StreamsMetricsImpl streamsMetrics,
final ThreadCache cache) {
if (!partitionsByInputTopic.isEmpty()) {
- final MockConsumer consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
consumer.assign(partitionsByInputTopic.values());
final Map startOffsets = new HashMap<>();
for (final TopicPartition topicPartition : partitionsByInputTopic.values()) {
@@ -439,15 +451,15 @@ private void setupTask(final StreamsConfig streamsConfig,
createRestoreConsumer(processorTopology.storeToChangelogTopic()),
stateRestoreListener),
processorTopology.storeToChangelogTopic(),
- new HashSet<>(partitionsByInputTopic.values()));
+ new HashSet<>(partitionsByInputTopic.values())
+ );
final RecordCollector recordCollector = new RecordCollectorImpl(
logContext,
TASK_ID,
- consumer,
- new StreamsProducer(producer, eosEnabled, logContext, streamsConfig.getString(StreamsConfig.APPLICATION_ID_CONFIG)),
+ testDriverProducer,
streamsConfig.defaultProductionExceptionHandler(),
- eosEnabled,
- streamsMetrics);
+ streamsMetrics
+ );
task = new StreamTask(
TASK_ID,
new HashSet<>(partitionsByInputTopic.values()),
@@ -459,7 +471,8 @@ private void setupTask(final StreamsConfig streamsConfig,
cache,
mockWallClockTime,
stateManager,
- recordCollector);
+ recordCollector
+ );
task.initializeIfNeeded();
task.completeRestoration();
((InternalProcessorContext) task.context()).setRecordContext(new ProcessorRecordContext(
@@ -467,7 +480,8 @@ private void setupTask(final StreamsConfig streamsConfig,
-1L,
-1,
ProcessorContextImpl.NONEXIST_TOPIC,
- new RecordHeaders()));
+ new RecordHeaders())
+ );
} else {
task = null;
}
@@ -497,7 +511,8 @@ public void pipeInput(final ConsumerRecord consumerRecord) {
consumerRecord.timestamp(),
consumerRecord.key(),
consumerRecord.value(),
- consumerRecord.headers());
+ consumerRecord.headers()
+ );
}
private void pipeRecord(final String topicName,
@@ -539,7 +554,8 @@ private void enqueueTaskRecord(final String inputTopic,
value == null ? ConsumerRecord.NULL_SIZE : value.length,
key,
value,
- headers)));
+ headers))
+ );
}
private void completeAllProcessableWork() {
@@ -557,7 +573,8 @@ private void completeAllProcessableWork() {
// Process the record ...
task.process(mockWallClockTime.milliseconds());
task.maybePunctuateStreamTime();
- task.commit();
+ task.prepareCommit();
+ commit(task.committableOffsetsAndMetadata());
captureOutputsAndReEnqueueInternalResults();
}
if (task.hasRecordsQueued()) {
@@ -570,6 +587,14 @@ private void completeAllProcessableWork() {
}
}
+ private void commit(final Map offsets) {
+ if (eosEnabled) {
+ testDriverProducer.commitTransaction(offsets);
+ } else {
+ consumer.commitSync(offsets);
+ }
+ }
+
private void processGlobalRecord(final TopicPartition globalInputTopicPartition,
final long timestamp,
final byte[] key,
@@ -586,7 +611,8 @@ private void processGlobalRecord(final TopicPartition globalInputTopicPartition,
value == null ? ConsumerRecord.NULL_SIZE : value.length,
key,
value,
- headers));
+ headers)
+ );
globalStateTask.flushState();
}
@@ -637,7 +663,8 @@ private void captureOutputsAndReEnqueueInternalResults() {
record.timestamp(),
record.key(),
record.value(),
- record.headers());
+ record.headers()
+ );
}
if (globalInputTopicPartition != null) {
@@ -646,7 +673,8 @@ private void captureOutputsAndReEnqueueInternalResults() {
record.timestamp(),
record.key(),
record.value(),
- record.headers());
+ record.headers()
+ );
}
}
}
@@ -691,7 +719,8 @@ public void advanceWallClockTime(final Duration advance) {
mockWallClockTime.sleep(advance.toMillis());
if (task != null) {
task.maybePunctuateSystemTime();
- task.commit();
+ task.prepareCommit();
+ commit(task.committableOffsetsAndMetadata());
}
completeAllProcessableWork();
}
@@ -1096,7 +1125,8 @@ public SessionStore getSessionStore(final String name) {
*/
public void close() {
if (task != null) {
- task.closeClean();
+ final Map checkpoint = task.prepareCloseClean();
+ task.closeClean(checkpoint);
}
if (globalStateTask != null) {
try {
diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/processor/internals/TestDriverProducer.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/processor/internals/TestDriverProducer.java
new file mode 100644
index 0000000000000..a33b51962bd69
--- /dev/null
+++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/processor/internals/TestDriverProducer.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.utils.LogContext;
+
+import java.util.Map;
+
+public class TestDriverProducer extends StreamsProducer {
+
+ public TestDriverProducer(final Producer producer,
+ final boolean eosEnabled,
+ final String applicationId,
+ final LogContext logContext) {
+ super(producer, eosEnabled, applicationId, logContext);
+ }
+
+ public void commitTransaction(final Map offsets) throws ProducerFencedException {
+ super.commitTransaction(offsets);
+ }
+}