diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index c586fb41be414..837fe29607c47 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -78,6 +78,7 @@ import java.util.Map; import java.util.Objects; import java.util.Properties; +import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -618,11 +619,22 @@ private static int parseAcks(String acksString) { public void initTransactions() { throwIfNoTransactionManager(); throwIfProducerClosed(); + maybeAllocateTransactionalId(); + TransactionalRequestResult result = transactionManager.initializeTransactions(); sender.wakeup(); result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS); } + private void maybeAllocateTransactionalId() { + String allocatedTransactionalId = transactionManager.transactionalId(); + if (allocatedTransactionalId == null || allocatedTransactionalId.isEmpty()) { + String threadProducerId = "thread-producer-" + UUID.randomUUID().toString(); + log.info("Allocating thread producer id: {}", threadProducerId); + transactionManager.setTransactionalId(threadProducerId); + } + } + /** * Should be called before the start of each new transaction. Note that prior to the first invocation * of this method, you must invoke {@link #initTransactions()} exactly one time. diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java index cd091542dc8db..713e9e4a7662b 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java @@ -83,7 +83,7 @@ public class TransactionManager { private static final int NO_LAST_ACKED_SEQUENCE_NUMBER = -1; private final Logger log; - private final String transactionalId; + private String transactionalId; private final int transactionTimeoutMs; private static class TopicPartitionBookkeeper { @@ -272,6 +272,10 @@ public TransactionManager(LogContext logContext, String transactionalId, int tra this(new LogContext(), null, 0, 100L); } + public void setTransactionalId(String transactionalId) { + this.transactionalId = transactionalId; + } + public synchronized TransactionalRequestResult initializeTransactions() { return handleCachedTransactionRequestResult(() -> { transitionTo(State.INITIALIZING); @@ -986,8 +990,9 @@ private TxnOffsetCommitHandler txnOffsetCommitHandler(TransactionalRequestResult offsetAndMetadata.metadata(), offsetAndMetadata.leaderEpoch()); pendingTxnOffsetCommits.put(entry.getKey(), committedOffset); } + TxnOffsetCommitRequest.Builder builder = new TxnOffsetCommitRequest.Builder(transactionalId, consumerGroupId, - producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, pendingTxnOffsetCommits); + producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, pendingTxnOffsetCommits); return new TxnOffsetCommitHandler(result, builder); } diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java index f08eecaac4743..455f8722256fd 100644 --- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java +++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java @@ -263,6 +263,13 @@ public class StreamsConfig extends AbstractConfig { @SuppressWarnings("WeakerAccess") public static final String EXACTLY_ONCE = "exactly_once"; + /** + * Config to use thread level producer + */ + @SuppressWarnings("WeakerAccess") + public static final String USE_EOS_THREAD_PRODUCER_CONFIG = "use.eos.thread.producer"; + private static final String USE_EOS_THREAD_PRODUCER_DOC = "Flag to use thread level producer."; + /** {@code application.id} */ @SuppressWarnings("WeakerAccess") public static final String APPLICATION_ID_CONFIG = "application.id"; @@ -568,6 +575,11 @@ public class StreamsConfig extends AbstractConfig { in(NO_OPTIMIZATION, OPTIMIZE), Importance.MEDIUM, TOPOLOGY_OPTIMIZATION_DOC) + .define(USE_EOS_THREAD_PRODUCER_CONFIG, + Type.BOOLEAN, + false, + Importance.MEDIUM, + USE_EOS_THREAD_PRODUCER_DOC) // LOW diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStandbyTasks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStandbyTasks.java index a99e45147b9ec..1d607b21be9a1 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStandbyTasks.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStandbyTasks.java @@ -21,7 +21,7 @@ class AssignedStandbyTasks extends AssignedTasks { AssignedStandbyTasks(final LogContext logContext) { - super(logContext, "standby task"); + super(logContext, "standby task", null, null, "dummy-group-id"); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java index f0019ec96722c..4d44f97f0fcdd 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasks.java @@ -16,9 +16,11 @@ */ package org.apache.kafka.streams.processor.internals; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; import org.apache.kafka.streams.errors.TaskMigratedException; import org.apache.kafka.streams.processor.TaskId; @@ -35,9 +37,14 @@ class AssignedStreamsTasks extends AssignedTasks implements Restorin private final Map restoring = new HashMap<>(); private final Set restoredPartitions = new HashSet<>(); private final Map restoringByPartition = new HashMap<>(); - - AssignedStreamsTasks(final LogContext logContext) { - super(logContext, "stream task"); + private final Producer threadProducer; + + AssignedStreamsTasks(final LogContext logContext, + final Producer threadProducer, + final Time time, + final String consumerGroupId) { + super(logContext, "stream task", threadProducer, time, consumerGroupId); + this.threadProducer = threadProducer; } @Override @@ -136,41 +143,8 @@ void addToRestoring(final StreamTask task) { * or if the task producer got fenced (EOS) */ int maybeCommitPerUserRequested() { - int committed = 0; - RuntimeException firstException = null; - - for (final Iterator it = running().iterator(); it.hasNext(); ) { - final StreamTask task = it.next(); - try { - if (task.commitRequested() && task.commitNeeded()) { - task.commit(); - committed++; - log.debug("Committed active task {} per user request in", task.id()); - } - } catch (final TaskMigratedException e) { - log.info("Failed to commit {} since it got migrated to another thread already. " + - "Closing it as zombie before triggering a new rebalance.", task.id()); - final RuntimeException fatalException = closeZombieTask(task); - if (fatalException != null) { - throw fatalException; - } - it.remove(); - throw e; - } catch (final RuntimeException t) { - log.error("Failed to commit StreamTask {} due to the following error:", - task.id(), - t); - if (firstException == null) { - firstException = t; - } - } - } - - if (firstException != null) { - throw firstException; - } - - return committed; + return commitInternal(log, threadProducer, + task -> task.commitRequested() && task.commitNeeded()); } /** diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java index e1bfe37247267..40fa3fd08c9c2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java @@ -16,8 +16,11 @@ */ 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.utils.LogContext; +import org.apache.kafka.common.utils.Time; import org.apache.kafka.streams.errors.LockException; import org.apache.kafka.streams.errors.StreamsException; import org.apache.kafka.streams.errors.TaskMigratedException; @@ -42,15 +45,24 @@ abstract class AssignedTasks { private final Map created = new HashMap<>(); private final Map suspended = new HashMap<>(); private final Set previousActiveTasks = new HashSet<>(); + private final Producer eosProducer; + private final Time time; + private final String consumerGroupId; // IQ may access this map. final Map running = new ConcurrentHashMap<>(); private final Map runningByPartition = new HashMap<>(); AssignedTasks(final LogContext logContext, - final String taskTypeName) { + final String taskTypeName, + final Producer eosProducer, + final Time time, + final String consumerGroupId) { this.taskTypeName = taskTypeName; this.log = logContext.logger(getClass()); + this.eosProducer = eosProducer; + this.time = time; + this.consumerGroupId = consumerGroupId; } void addNewTask(final T task) { @@ -276,13 +288,36 @@ Set previousTaskIds() { * or if the task producer got fenced (EOS) */ int commit() { + return commitInternal( + log, + eosProducer, + Task::commitNeeded + ); + } + + public interface TaskStatus { + boolean needsCommit(Task task); + } + + protected int commitInternal(final Logger log, + final Producer eosProducer, + final TaskStatus taskStatus) { int committed = 0; RuntimeException firstException = null; + + final Map pendingOffsets = new HashMap<>(); + final List externalCommitTasks = new ArrayList<>(); + for (final Iterator it = running().iterator(); it.hasNext(); ) { final T task = it.next(); try { - if (task.commitNeeded()) { - task.commit(); + if (taskStatus.needsCommit(task)) { + if (eosProducer != null) { + pendingOffsets.putAll(task.getPendingOffsets()); + externalCommitTasks.add(task); + } else { + task.commit(); + } committed++; } } catch (final TaskMigratedException e) { @@ -296,9 +331,7 @@ int commit() { throw e; } catch (final RuntimeException t) { log.error("Failed to commit {} {} due to the following error:", - taskTypeName, - task.id(), - t); + taskTypeName, task.id(), t); if (firstException == null) { firstException = t; } @@ -309,6 +342,17 @@ int commit() { throw firstException; } + if (!pendingOffsets.isEmpty()) { + final long startNs = time.nanoseconds(); + eosProducer.sendOffsetsToTransaction(pendingOffsets, consumerGroupId); + eosProducer.commitTransaction(); + final long commitLatency = time.nanoseconds() - startNs; + for (final Task task : externalCommitTasks) { + task.markCommitDone(commitLatency); + } + eosProducer.beginTransaction(); + } + return committed; } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index 836330f242617..1ffa77929595d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -82,6 +82,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator private Sensor closeTaskSensor; private long idleStartTime; private Producer producer; + private final boolean isTaskProducer; private boolean commitRequested = false; private boolean transactionInFlight = false; @@ -151,8 +152,9 @@ public StreamTask(final TaskId id, final StateDirectory stateDirectory, final ThreadCache cache, final Time time, - final ProducerSupplier producerSupplier) { - this(id, partitions, topology, consumer, changelogReader, config, metrics, stateDirectory, cache, time, producerSupplier, null); + final ProducerSupplier producerSupplier, + final boolean isTaskProducer) { + this(id, partitions, topology, consumer, changelogReader, config, metrics, stateDirectory, cache, time, producerSupplier, null, isTaskProducer); } public StreamTask(final TaskId id, @@ -166,9 +168,11 @@ public StreamTask(final TaskId id, final ThreadCache cache, final Time time, final ProducerSupplier producerSupplier, - final RecordCollector recordCollector) { + final RecordCollector recordCollector, + final boolean isTaskProducer) { super(id, partitions, topology, consumer, changelogReader, false, stateDirectory, config); + this.isTaskProducer = isTaskProducer; this.time = time; this.producerSupplier = producerSupplier; this.producer = producerSupplier.get(); @@ -179,14 +183,22 @@ public StreamTask(final TaskId id, final ProductionExceptionHandler productionExceptionHandler = config.defaultProductionExceptionHandler(); if (recordCollector == null) { + log.info("record collector is initialized on task"); this.recordCollector = new RecordCollectorImpl( id.toString(), logContext, productionExceptionHandler, ThreadMetrics.skipRecordSensor(streamsMetrics)); } else { + log.info("record collector given is non-null"); this.recordCollector = recordCollector; } + + if (this.producer == null) { + log.info("Initializing the record collector with producer null"); + } else { + log.info("Initializing the record collector with non-null producer"); + } this.recordCollector.init(this.producer); streamTimePunctuationQueue = new PunctuationQueue(); @@ -228,7 +240,7 @@ public StreamTask(final TaskId id, // initialize transactions if eos is turned on, which will block if the previous transaction has not // completed yet; do not start the first transaction until the topology has been initialized later - if (eosEnabled) { + if (isTaskProducer) { initializeTransactions(); } } @@ -252,7 +264,7 @@ public boolean initializeStateStores() { public void initializeTopology() { initTopology(); - if (eosEnabled) { + if (isTaskProducer) { try { this.producer.beginTransaction(); } catch (final ProducerFencedException fatal) { @@ -279,11 +291,13 @@ public void initializeTopology() { public void resume() { log.debug("Resuming"); if (eosEnabled) { - if (producer != null) { - throw new IllegalStateException("Task producer should be null."); - } +// if (producer != null) { +// throw new IllegalStateException("Task producer should be null."); +// } producer = producerSupplier.get(); - initializeTransactions(); + if (isTaskProducer) { + initializeTransactions(); + } recordCollector.init(producer); try { @@ -455,16 +469,10 @@ void commit(final boolean startNewTransaction) { stateMgr.checkpoint(activeTaskCheckpointableOffsets()); } - final Map consumedOffsetsAndMetadata = new HashMap<>(consumedOffsets.size()); - for (final Map.Entry entry : consumedOffsets.entrySet()) { - final TopicPartition partition = entry.getKey(); - final long offset = entry.getValue() + 1; - consumedOffsetsAndMetadata.put(partition, new OffsetAndMetadata(offset)); - stateMgr.putOffsetLimit(partition, offset); - } + final Map consumedOffsetsAndMetadata = getPendingOffsets(); try { - if (eosEnabled) { + if (isTaskProducer) { producer.sendOffsetsToTransaction(consumedOffsetsAndMetadata, applicationId); producer.commitTransaction(); transactionInFlight = false; @@ -479,9 +487,26 @@ void commit(final boolean startNewTransaction) { throw new TaskMigratedException(this, error); } + markCommitDone(time.nanoseconds() - startNs); + } + + @Override + public Map getPendingOffsets() { + final Map consumedOffsetsAndMetadata = new HashMap<>(consumedOffsets.size()); + for (final Map.Entry entry : consumedOffsets.entrySet()) { + final TopicPartition partition = entry.getKey(); + final long offset = entry.getValue() + 1; + consumedOffsetsAndMetadata.put(partition, new OffsetAndMetadata(offset)); + stateMgr.putOffsetLimit(partition, offset); + } + return consumedOffsetsAndMetadata; + } + + @Override + public void markCommitDone(final long commitLatency) { commitNeeded = false; commitRequested = false; - taskMetrics.taskCommitTimeSensor.record(time.nanoseconds() - startNs); + taskMetrics.taskCommitTimeSensor.record(commitLatency); } @Override @@ -598,7 +623,7 @@ void suspend(final boolean clean, } private void maybeAbortTransactionAndCloseRecordCollector(final boolean isZombie) { - if (eosEnabled && !isZombie) { + if (isTaskProducer && !isZombie) { try { if (transactionInFlight) { producer.abortTransaction(); @@ -616,7 +641,7 @@ private void maybeAbortTransactionAndCloseRecordCollector(final boolean isZombie } } - if (eosEnabled) { + if (isTaskProducer) { try { recordCollector.close(); } catch (final Throwable e) { @@ -844,7 +869,8 @@ void requestCommit() { /** * Whether or not a request has been made to commit the current state */ - boolean commitRequested() { + @Override + public boolean commitRequested() { return commitRequested; } 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 d3efa9e0e05de..bf219f4df94fb 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 @@ -29,6 +29,7 @@ import org.apache.kafka.common.Metric; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.serialization.ByteArrayDeserializer; @@ -61,6 +62,7 @@ import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; +import static java.lang.String.format; import static java.util.Collections.singleton; public class StreamThread extends Thread { @@ -391,7 +393,6 @@ Collection createTasks(final Consumer consumer, log.trace("Created task {} with assigned partitions {}", taskId, partitions); createdTasks.add(task); } - } return createdTasks; } @@ -430,6 +431,11 @@ static class TaskCreator extends AbstractTaskCreator { this.cache = cache; this.clientSupplier = clientSupplier; this.threadProducer = threadProducer; + if (threadProducer == null) { + log.info("Initializing with thread producer to null in task creator"); + } else { + log.info("Initializing with thread producer to non-null in task creator"); + } this.threadClientId = threadClientId; createTaskSensor = ThreadMetrics.createTaskSensor(streamsMetrics); } @@ -439,7 +445,7 @@ StreamTask createTask(final Consumer consumer, final TaskId taskId, final Set partitions) { createTaskSensor.record(); - + log.info("creating task {}", taskId); return new StreamTask( taskId, partitions, @@ -451,7 +457,8 @@ StreamTask createTask(final Consumer consumer, stateDirectory, cache, time, - () -> createProducer(taskId)); + () -> createProducer(taskId), + threadProducer == null); } private Producer createProducer(final TaskId id) { @@ -563,6 +570,8 @@ StandbyTask createTask(final Consumer consumer, final Consumer consumer; final InternalTopologyBuilder builder; + final boolean eosEnabled; + public static StreamThread create(final InternalTopologyBuilder builder, final StreamsConfig config, final KafkaClientSupplier clientSupplier, @@ -588,18 +597,35 @@ public static StreamThread create(final InternalTopologyBuilder builder, final Duration pollTime = Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG)); final StoreChangelogReader changelogReader = new StoreChangelogReader(restoreConsumer, pollTime, userStateRestoreListener, logContext); - Producer threadProducer = null; final boolean eosEnabled = StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)); - if (!eosEnabled) { + + final boolean useEosThreadProducer = config.getBoolean(StreamsConfig.USE_EOS_THREAD_PRODUCER_CONFIG); + + final Producer threadProducer; + final String applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); + if (useEosThreadProducer || !eosEnabled) { final Map producerConfigs = config.getProducerConfigs(getThreadProducerClientId(threadClientId)); log.info("Creating shared producer client"); + if (eosEnabled) { + producerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, applicationId + "-" + threadClientId); + } threadProducer = clientSupplier.getProducer(producerConfigs); + } else { + threadProducer = null; } + final Producer eosThreadProducer = eosEnabled ? threadProducer : null; + final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, threadClientId); final ThreadCache cache = new ThreadCache(logContext, cacheSizeBytes, streamsMetrics); + if (threadProducer == null) { + log.info("Init thread producer on stream-thread with null"); + } else { + log.info("Init thread producer on stream-thread with non-null"); + } + final AbstractTaskCreator activeTaskCreator = new TaskCreator( builder, config, @@ -620,6 +646,7 @@ public static StreamThread create(final InternalTopologyBuilder builder, changelogReader, time, log); + final TaskManager taskManager = new TaskManager( changelogReader, processId, @@ -629,11 +656,10 @@ public static StreamThread create(final InternalTopologyBuilder builder, activeTaskCreator, standbyTaskCreator, adminClient, - new AssignedStreamsTasks(logContext), + new AssignedStreamsTasks(logContext, eosThreadProducer, time, applicationId), new AssignedStandbyTasks(logContext)); log.info("Creating consumer client"); - final String applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); final Map consumerConfigs = config.getMainConsumerConfigs(applicationId, getConsumerClientId(threadClientId), threadIdx); consumerConfigs.put(StreamsConfig.InternalConfig.TASK_MANAGER_FOR_PARTITION_ASSIGNOR, taskManager); final AtomicInteger assignmentErrorCode = new AtomicInteger(); @@ -659,7 +685,8 @@ public static StreamThread create(final InternalTopologyBuilder builder, builder, threadClientId, logContext, - assignmentErrorCode) + assignmentErrorCode, + eosEnabled) .updateThreadMetadata(getSharedAdminClientId(clientId)); } @@ -674,7 +701,8 @@ public StreamThread(final Time time, final InternalTopologyBuilder builder, final String threadClientId, final LogContext logContext, - final AtomicInteger assignmentErrorCode) { + final AtomicInteger assignmentErrorCode, + final boolean eosEnabled) { super(threadClientId); this.stateLock = new Object(); @@ -715,6 +743,8 @@ public StreamThread(final Time time, this.commitTimeMs = config.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG); this.numIterations = 1; + + this.eosEnabled = eosEnabled; } private static final class InternalConsumerConfig extends ConsumerConfig { @@ -759,6 +789,7 @@ public void run() { } boolean cleanRun = false; try { + maybeInitializeTransactions(); runLoop(); cleanRun = true; } catch (final KafkaException e) { @@ -775,6 +806,28 @@ public void run() { } } + private void maybeInitializeTransactions() { + try { + // This is a thread-level txn producer + if (eosEnabled && producer != null) { + producer.initTransactions(); + producer.beginTransaction(); + } + } catch (final TimeoutException retriable) { + log.error( + "Timeout exception caught when initializing transactions for current stream thread. " + + "This might happen if the broker is slow to respond, if the network connection to " + + "the broker was interrupted, or if similar circumstances arise. " + + "You can increase producer parameter `max.block.ms` to increase this timeout.", + retriable + ); + throw new StreamsException( + format("%sFailed to initialize stream thread due to timeout.", logPrefix), + retriable + ); + } + } + private void setRebalanceException(final Throwable rebalanceException) { this.rebalanceException = rebalanceException; } 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 812e7e1131ca2..3295ab05ec77e 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 @@ -16,6 +16,7 @@ */ package org.apache.kafka.streams.processor.internals; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.streams.errors.StreamsException; import org.apache.kafka.streams.processor.ProcessorContext; @@ -23,6 +24,8 @@ import org.apache.kafka.streams.processor.TaskId; import java.util.Collection; +import java.util.Collections; +import java.util.Map; import java.util.Set; public interface Task { @@ -36,10 +39,22 @@ public interface Task { boolean commitNeeded(); + default boolean commitRequested() { + return false; + } + void initializeTopology(); void commit(); + default Map getPendingOffsets() { + return Collections.emptyMap(); + } + + default void markCommitDone(long commitLatency) { + // No-op + } + void suspend(); void resume(); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java index ffd0f8baae6b9..d07597c35337c 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java @@ -19,6 +19,7 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.errors.TaskMigratedException; import org.apache.kafka.streams.processor.TaskId; @@ -52,7 +53,7 @@ public class AssignedStreamsTasksTest { @Before public void before() { - assignedTasks = new AssignedStreamsTasks(new LogContext("log ")); + assignedTasks = new AssignedStreamsTasks(new LogContext("log"), null, new MockTime(), "dummy-group-id"); EasyMock.expect(t1.id()).andReturn(taskId1).anyTimes(); EasyMock.expect(t2.id()).andReturn(taskId2).anyTimes(); } @@ -451,6 +452,14 @@ public void shouldReturnNumberOfPunctuations() { EasyMock.verify(t1); } + @Test + public void shouldLetThreadProducerCommit() { +// final Map producerConfig = new HashMap<>(); +// +// assignedTasks = new AssignedStreamsTasks(new LogContext("log"), new DefaultKafkaClientSupplier().getProducer(), new MockTime(), "dummy-group-id"); + + } + private void addAndInitTask() { assignedTasks.addNewTask(t1); assignedTasks.initializeNewTasks(); 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 a70c10c3a7216..836c231ee479d 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 @@ -246,8 +246,8 @@ public void initTransactions() { throw new TimeoutException("test"); } }, - null - ); + null, + false); fail("Expected an exception"); } catch (final StreamsException expected) { // make sure we log the explanation as an ERROR @@ -300,8 +300,8 @@ public void initTransactions() { } } }, - null - ); + null, + false); testTask.initializeTopology(); testTask.suspend(); timeOut.set(true); @@ -849,7 +849,7 @@ public void shouldFlushRecordCollectorOnFlushState() { public void flush() { flushed.set(true); } - }); + }, false); streamTask.flushState(); assertTrue(flushed.get()); } @@ -1424,7 +1424,7 @@ public void shouldReturnOffsetsForRepartitionTopicsForPurging() { stateDirectory, null, time, - () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer)); + () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer), false); task.initializeStateStores(); task.initializeTopology(); @@ -1496,7 +1496,7 @@ private StreamTask createStatefulTask(final StreamsConfig config, final boolean stateDirectory, null, time, - () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer)); + () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer), false); } private StreamTask createStatefulTaskThatThrowsExceptionOnClose() { @@ -1506,18 +1506,19 @@ private StreamTask createStatefulTaskThatThrowsExceptionOnClose() { singletonList(stateStore), Collections.emptyMap()); + final boolean enableEOS = true; return new StreamTask( taskId00, partitions, topology, consumer, changelogReader, - createConfig(true), + createConfig(enableEOS), streamsMetrics, stateDirectory, null, time, - () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer)); + () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer), enableEOS); } private StreamTask createStatelessTask(final StreamsConfig streamsConfig) { @@ -1531,6 +1532,8 @@ private StreamTask createStatelessTask(final StreamsConfig streamsConfig) { source1.addChild(processorSystemTime); source2.addChild(processorSystemTime); + final boolean enableEOS = StreamsConfig.EXACTLY_ONCE.equals(streamsConfig.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)); + return new StreamTask( taskId00, partitions, @@ -1542,7 +1545,7 @@ private StreamTask createStatelessTask(final StreamsConfig streamsConfig) { stateDirectory, null, time, - () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer)); + () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer), enableEOS); } // this task will throw exception when processing (on partition2), flushing, suspending and closing @@ -1568,7 +1571,7 @@ private StreamTask createTaskThatThrowsException(final boolean enableEos) { stateDirectory, null, time, - () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer)) { + () -> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer), false) { @Override protected void flushState() { throw new RuntimeException("KABOOM!"); 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 aff5d6c74861c..25c42272c2d1a 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 @@ -106,6 +106,7 @@ import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -353,8 +354,8 @@ public void shouldNotCommitBeforeTheCommitInterval() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ); + new AtomicInteger(), + false); thread.setNow(mockTime.milliseconds()); thread.maybeCommit(); mockTime.sleep(commitInterval - 10L); @@ -452,7 +453,53 @@ public void shouldRespectNumIterationsInMainLoop() { thread.runOnce(); assertThat(thread.currentNumIterations(), equalTo(1)); + } + + @Test + public void threadProducerOnEosShouldCommit() { + final MockProcessor mockProcessor = new MockProcessor(PunctuationType.WALL_CLOCK_TIME, 10L); + internalTopologyBuilder.addSource(null, "source1", null, null, null, topic1); + internalTopologyBuilder.addProcessor("processor1", () -> mockProcessor, "source1"); + internalTopologyBuilder.addProcessor("processor2", () -> new MockProcessor(PunctuationType.STREAM_TIME, 10L), "source1"); + + final Properties properties = new Properties(); + properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L); + properties.put(StreamsConfig.USE_EOS_THREAD_PRODUCER_CONFIG, true); + properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE); + final StreamsConfig config = new StreamsConfig(StreamsTestUtils.getStreamsConfig(applicationId, + "localhost:2171", + Serdes.ByteArraySerde.class.getName(), + Serdes.ByteArraySerde.class.getName(), + properties)); + final StreamThread thread = createStreamThread(clientId, config, true); + + thread.setState(StreamThread.State.STARTING); + thread.setState(StreamThread.State.PARTITIONS_REVOKED); + final Set assignedPartitions = Collections.singleton(t1p1); + thread.taskManager().setAssignmentMetadata( + Collections.singletonMap( + new TaskId(0, t1p1.partition()), + assignedPartitions), + Collections.emptyMap()); + + final MockConsumer mockConsumer = (MockConsumer) thread.consumer; + mockConsumer.assign(Collections.singleton(t1p1)); + mockConsumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L)); + thread.rebalanceListener.onPartitionsAssigned(assignedPartitions); + thread.runOnce(); + + // processed one record, punctuated after the first record, and hence num.iterations is still 1 + long offset = -1; + addRecord(mockConsumer, ++offset, 0L); + thread.runOnce(); + + assertThat(thread.currentNumIterations(), equalTo(1)); + + mockProcessor.requestCommit(); + addRecord(mockConsumer, ++offset, 15L); + // Mock producer should throw because it's not enabled for transactional support but wanted. + assertThrows(IllegalStateException.class, () -> thread.runOnce()); } @Test @@ -479,8 +526,8 @@ public void shouldNotCauseExceptionIfNothingCommitted() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ); + new AtomicInteger(), + false); thread.setNow(mockTime.milliseconds()); thread.maybeCommit(); mockTime.sleep(commitInterval - 10L); @@ -514,8 +561,8 @@ public void shouldCommitAfterTheCommitInterval() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ); + new AtomicInteger(), + false); thread.setNow(mockTime.milliseconds()); thread.maybeCommit(); mockTime.sleep(commitInterval + 1); @@ -664,8 +711,8 @@ public void shouldShutdownTaskManagerOnClose() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ).updateThreadMetadata(getSharedAdminClientId(clientId)); + new AtomicInteger(), + false).updateThreadMetadata(getSharedAdminClientId(clientId)); thread.setStateListener( (t, newState, oldState) -> { if (oldState == StreamThread.State.CREATED && newState == StreamThread.State.STARTING) { @@ -697,8 +744,8 @@ public void shouldShutdownTaskManagerOnCloseWithoutStart() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ).updateThreadMetadata(getSharedAdminClientId(clientId)); + new AtomicInteger(), + false).updateThreadMetadata(getSharedAdminClientId(clientId)); thread.shutdown(); EasyMock.verify(taskManager); } @@ -751,7 +798,7 @@ private void setStreamThread(final StreamThread streamThread) { null, null, null, - new AssignedStreamsTasks(new LogContext()), + new AssignedStreamsTasks(new LogContext(), null, new MockTime(), applicationId), new AssignedStandbyTasks(new LogContext())); taskManager.setConsumer(mockStreamThreadConsumer); taskManager.setAssignmentMetadata(Collections.emptyMap(), Collections.emptyMap()); @@ -769,8 +816,8 @@ private void setStreamThread(final StreamThread streamThread) { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ).updateThreadMetadata(getSharedAdminClientId(clientId)); + new AtomicInteger(), + false).updateThreadMetadata(getSharedAdminClientId(clientId)); mockStreamThreadConsumer.setStreamThread(thread); mockStreamThreadConsumer.assign(assignedPartitions); @@ -803,8 +850,8 @@ public void shouldOnlyShutdownOnce() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ).updateThreadMetadata(getSharedAdminClientId(clientId)); + new AtomicInteger(), + false).updateThreadMetadata(getSharedAdminClientId(clientId)); thread.shutdown(); // Execute the run method. Verification of the mock will check that shutdown was only done once thread.run(); @@ -1634,8 +1681,8 @@ public void producerMetricsVerificationWithoutEOS() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ); + new AtomicInteger(), + false); final MetricName testMetricName = new MetricName("test_metric", "", "", new HashMap<>()); final Metric testMetric = new KafkaMetric( new Object(), @@ -1673,8 +1720,8 @@ public void adminClientMetricsVerification() { internalTopologyBuilder, clientId, new LogContext(""), - new AtomicInteger() - ); + new AtomicInteger(), + false); final MetricName testMetricName = new MetricName("test_metric", "", "", new HashMap<>()); final Metric testMetric = new KafkaMetric( new Object(), 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 f48c31c1d91c2..5780ddd6c19e3 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 @@ -317,7 +317,7 @@ private StreamTask createStreamsTask(final StreamsConfig streamsConfig, stateDirectory, null, new MockTime(), - () -> clientSupplier.getProducer(new HashMap<>())) { + () -> clientSupplier.getProducer(new HashMap<>()), false) { @Override protected void updateOffsetLimits() {} }; 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 2b1428f9f8968..5370c3906caf9 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 @@ -374,7 +374,7 @@ public void onRestoreEnd(final TopicPartition topicPartition, final String store stateDirectory, cache, mockWallClockTime, - () -> producer); + () -> producer, false); task.initializeStateStores(); task.initializeTopology(); ((InternalProcessorContext) task.context()).setRecordContext(new ProcessorRecordContext( diff --git a/tests/kafkatest/tests/streams/base_streams_test.py b/tests/kafkatest/tests/streams/base_streams_test.py index 53e4231edb419..2b2e92fdb6ec1 100644 --- a/tests/kafkatest/tests/streams/base_streams_test.py +++ b/tests/kafkatest/tests/streams/base_streams_test.py @@ -67,7 +67,7 @@ def assert_produce(self, topic, test_state, num_messages=5, timeout_sec=60): wait_until(lambda: producer.num_acked >= num_messages, timeout_sec=timeout_sec, - err_msg="At %s failed to send messages " % test_state) + err_msg="At %s failed to send messages. Expected: %s, actual: %s " % (test_state, num_messages, producer.num_acked)) def assert_consume(self, client_id, test_state, topic, num_messages=5, timeout_sec=60): consumer = self.get_consumer(client_id, topic, num_messages) @@ -75,7 +75,7 @@ def assert_consume(self, client_id, test_state, topic, num_messages=5, timeout_s wait_until(lambda: consumer.total_consumed() >= num_messages, timeout_sec=timeout_sec, - err_msg="At %s streams did not process messages in %s seconds " % (test_state, timeout_sec)) + err_msg="At %s streams did not process messages in %s seconds. Expected: %s, actual: %s " % (test_state, timeout_sec, num_messages, consumer.total_consumed())) @staticmethod def get_configs(extra_configs=""):