Skip to content

KAFKA-9441: Unify committing within TaskManager #8218

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 6 commits into from
Mar 19, 2020
+1,467 −706
Merged

KAFKA-9441: Unify committing within TaskManager #8218

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b5de5b1
KAFKA-9441: Unify committing within TaskManager
mjsax Mar 4, 2020
2002144
Github comments
mjsax Mar 16, 2020
a2f2c04
Github comments
mjsax Mar 17, 2020
45e5633
Fix bug
mjsax Mar 17, 2020
693b33d
Github comments
mjsax Mar 17, 2020
4b03a48
Bug fix
mjsax Mar 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@
files="StreamsPartitionAssignor.java"/>

<suppress checks="NPathComplexity"
files="(ProcessorStateManager|InternalTopologyBuilder|StreamsPartitionAssignor|StreamThread).java"/>
files="(ProcessorStateManager|InternalTopologyBuilder|StreamsPartitionAssignor|StreamThread|TaskManager).java"/>

<suppress checks="(FinalLocalVariable|UnnecessaryParentheses|BooleanExpressionComplexity|CyclomaticComplexity|WhitespaceAfter|LocalVariableName)"
files="Murmur3.java"/>
Expand Down
68 changes: 42 additions & 26 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,20 +47,20 @@
import static org.apache.kafka.streams.StreamsConfig.EXACTLY_ONCE;

class ActiveTaskCreator {
private final String applicationId;
private final InternalTopologyBuilder builder;
private final StreamsConfig config;
private final StreamsMetricsImpl streamsMetrics;
private final StateDirectory stateDirectory;
private final ChangelogReader storeChangelogReader;
private final Time time;
private final Logger log;
private final String threadId;
private final ThreadCache cache;
private final Producer<byte[], byte[]> threadProducer;
private final Time time;
private final KafkaClientSupplier clientSupplier;
private final Map<TaskId, Producer<byte[], byte[]>> taskProducers;
private final String threadId;
private final Logger log;
private final Sensor createTaskSensor;
private final String applicationId;
private final Producer<byte[], byte[]> threadProducer;
private final Map<TaskId, StreamsProducer> taskProducers;

private static String getThreadProducerClientId(final String threadClientId) {
return threadClientId + "-producer";
Expand All @@ -80,32 +80,44 @@ private static String getTaskProducerClientId(final String threadClientId, final
final KafkaClientSupplier clientSupplier,
final String threadId,
final Logger log) {
applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);
this.builder = builder;
this.config = config;
this.streamsMetrics = streamsMetrics;
this.stateDirectory = stateDirectory;
this.storeChangelogReader = storeChangelogReader;
this.cache = cache;
this.time = time;
this.clientSupplier = clientSupplier;
this.threadId = threadId;
this.log = log;

createTaskSensor = ThreadMetrics.createTaskSensor(threadId, streamsMetrics);
applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);

if (EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG))) {
threadProducer = null;
taskProducers = new HashMap<>();
} else {
log.info("Creating thread producer client");

@abbccdda abbccdda Mar 11, 2020

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: we could log thread-id here for easier log search.

@mjsax mjsax Mar 12, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The threadId is already added to the log prefix when the log object is created in StreamsThread


final String threadProducerClientId = getThreadProducerClientId(threadId);
final Map<String, Object> producerConfigs = config.getProducerConfigs(threadProducerClientId);
log.info("Creating thread producer client");

threadProducer = clientSupplier.getProducer(producerConfigs);
taskProducers = Collections.emptyMap();
}
}

StreamsProducer streamsProducerForTask(final TaskId taskId) {

@abbccdda abbccdda Mar 16, 2020

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need a unit test for this function.

@guozhangwang guozhangwang Mar 16, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the next PR (all other comments with this tag means no changes required for this PR): my understanding is that we would make the thread-producer also a StreamsProducer instead of a KafkaProducer which would be used to commitTransaction under eosBeta, is that right?

@mjsax mjsax Mar 17, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correct. For eos-beta there will be one StreamsProducer that is shared over all tasks.

if (threadProducer != null) {
throw new IllegalStateException("Producer per thread is used");
}

this.cache = cache;
this.threadId = threadId;
this.clientSupplier = clientSupplier;

createTaskSensor = ThreadMetrics.createTaskSensor(threadId, streamsMetrics);
final StreamsProducer taskProducer = taskProducers.get(taskId);
if (taskProducer == null) {
throw new IllegalStateException("Unknown TaskId: " + taskId);
}
return taskProducer;
}

Collection<Task> createTasks(final Consumer<byte[], byte[]> consumer,
Expand All @@ -132,23 +144,27 @@ Collection<Task> createTasks(final Consumer<byte[], byte[]> consumer,
partitions
);

final StreamsProducer streamsProducer;
if (threadProducer == null) {
final String taskProducerClientId = getTaskProducerClientId(threadId, taskId);
final Map<String, Object> producerConfigs = config.getProducerConfigs(taskProducerClientId);
producerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, applicationId + "-" + taskId);
log.info("Creating producer client for task {}", taskId);
taskProducers.put(taskId, clientSupplier.getProducer(producerConfigs));
streamsProducer = new StreamsProducer(
clientSupplier.getProducer(producerConfigs),
true,
applicationId,
logContext);
taskProducers.put(taskId, streamsProducer);
} else {
streamsProducer = new StreamsProducer(threadProducer, false, null, logContext);
}

final RecordCollector recordCollector = new RecordCollectorImpl(
logContext,
taskId,
consumer,
threadProducer != null ?
new StreamsProducer(threadProducer, false, logContext, applicationId) :
new StreamsProducer(taskProducers.get(taskId), true, logContext, applicationId),
streamsProducer,
config.defaultProductionExceptionHandler(),
EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)),
streamsMetrics
);

Expand Down Expand Up @@ -178,18 +194,18 @@ void closeThreadProducerIfNeeded() {
try {
threadProducer.close();
} catch (final RuntimeException e) {
throw new StreamsException("Thread Producer encounter unexpected error trying to close", e);
throw new StreamsException("Thread Producer encounter error trying to close", e);
}
}
}

void closeAndRemoveTaskProducerIfNeeded(final TaskId id) {
final Producer<byte[], byte[]> producer = taskProducers.remove(id);
if (producer != null) {
final StreamsProducer taskProducer = taskProducers.remove(id);
if (taskProducer != null) {
try {
producer.close();
taskProducer.kafkaProducer().close();

@guozhangwang guozhangwang Mar 16, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: we can have a wrapped StreamsProducer#close / metrics, and then #kafkaProducer would be for testing-only.

@mjsax mjsax Mar 17, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could, but the idea was that ActiveTaskCreator creates the producer via new KafkaProducer() and thus it should call KafkaProducer#close(), too, and not delegate it to StreamsProducer. Thoughts?

@guozhangwang guozhangwang Mar 17, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, I think moving forward we would create and maintain both the single thread-producer and task-producers as StreamsProducer objects right?

@mjsax mjsax Mar 17, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SGTM

} catch (final RuntimeException e) {
throw new StreamsException("[" + id + "] Producer encounter unexpected error trying to close", e);
throw new StreamsException("[" + id + "] Producer encounter error trying to close", e);
}
}
}
Expand All @@ -205,8 +221,8 @@ Map<MetricName, Metric> producerMetrics() {
// When EOS is turned on, each task will have its own producer client
// and the producer object passed in here will be null. We would then iterate through
// all the active tasks and add their metrics to the output metrics map.
for (final Map.Entry<TaskId, Producer<byte[], byte[]>> entry : taskProducers.entrySet()) {
final Map<MetricName, ? extends Metric> taskProducerMetrics = entry.getValue().metrics();
for (final Map.Entry<TaskId, StreamsProducer> entry : taskProducers.entrySet()) {
final Map<MetricName, ? extends Metric> taskProducerMetrics = entry.getValue().kafkaProducer().metrics();
result.putAll(taskProducerMetrics);
}
}
Expand Down
3 changes: 0 additions & 3 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
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.header.Headers;
Expand Down Expand Up @@ -45,8 +44,6 @@ <K, V> void send(final String topic,
final Serializer<V> valueSerializer,
final StreamPartitioner<? super K, ? super V> partitioner);

void commit(final Map<TopicPartition, OffsetAndMetadata> offsets);

@mjsax mjsax Mar 4, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We moved this to TaskManager

@guozhangwang guozhangwang Mar 12, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please see my other comment above --- I think it is cleaner to just call foreach(active-task) task.recordCollector.commit inside the task-manager; and inside RecordCollectorImpl we check that eosEnabled is always true, otherwise illegal-state thrown.

In the next PR where we have the thread-producer, we could then only create a single recordCollector object that is shared among all active tasks and wraps the thread-producer, and then the caller taskManager code then can just get one active task and call its record-collector's commit function knowing that is sufficient to actually commit for all tasks since everyone is using the same record-collector.

WDYT?

@guozhangwang guozhangwang Mar 16, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After syncing offline about this, I think I'm convinced now that moving this logic into TaskManager is better.


/**
* Initialize the internal {@link Producer}; note this function should be made idempotent
*
Expand Down
29 changes: 1 addition & 28 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,6 @@
*/
package org.apache.kafka.streams.processor.internals;

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.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
Expand All @@ -33,7 +30,6 @@
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.SecurityDisabledException;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.UnknownServerException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.metrics.Sensor;
Expand All @@ -59,7 +55,6 @@ public class RecordCollectorImpl implements RecordCollector {

private final Logger log;
private final TaskId taskId;
private final Consumer<byte[], byte[]> mainConsumer;
private final StreamsProducer streamsProducer;
private final ProductionExceptionHandler productionExceptionHandler;
private final Sensor droppedRecordsSensor;
Expand All @@ -73,17 +68,14 @@ public class RecordCollectorImpl implements RecordCollector {
*/
public RecordCollectorImpl(final LogContext logContext,
final TaskId taskId,
final Consumer<byte[], byte[]> mainConsumer,
final StreamsProducer streamsProducer,
final ProductionExceptionHandler productionExceptionHandler,
final boolean eosEnabled,
final StreamsMetricsImpl streamsMetrics) {
this.log = logContext.logger(getClass());
this.taskId = taskId;
this.mainConsumer = mainConsumer;
this.streamsProducer = streamsProducer;
this.productionExceptionHandler = productionExceptionHandler;
this.eosEnabled = eosEnabled;
this.eosEnabled = streamsProducer.eosEnabled();

final String threadId = Thread.currentThread().getName();
this.droppedRecordsSensor = TaskMetrics.droppedRecordsSensorOrSkippedRecordsSensor(threadId, taskId.toString(), streamsMetrics);
Expand Down Expand Up @@ -242,25 +234,6 @@ private boolean isFatalException(final Exception exception) {
return securityException || communicationException;
}

public void commit(final Map<TopicPartition, OffsetAndMetadata> offsets) {
if (eosEnabled) {
streamsProducer.commitTransaction(offsets);
} else {
try {
mainConsumer.commitSync(offsets);
} 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 for task " + taskId, error);
} catch (final KafkaException error) {
throw new StreamsException("Error encountered committing offsets via consumer for task " + taskId, error);
}
}

}

/**
* @throws StreamsException fatal error that should cause the thread to die
* @throws TaskMigratedException recoverable error that would cause the task to be removed
Expand Down
Loading