From 931fc4f610db5162b05497c8e0278494e8195b32 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Mon, 1 Mar 2021 11:03:34 -0500 Subject: [PATCH] KAFKA-10340: Proactively close producer when cancelling source tasks (#10016) Close the producer in `WorkerSourceTask` when the latter is cancelled. If the broker do not autocreate the topic, and the connector is not configured to create topics written by the source connector, then the `WorkerSourceTask` main thread will block forever until the topic is created, and will not stop if cancelled or scheduled for shutdown by the worker. Expanded an existing unit test for the WorkerSourceTask class to ensure that the producer is closed when the task is abandoned, and added a new integration test that guarantees that tasks are still shut down even when their producers are trying to write to topics that do not exist. Author: Chris Egerton Reviewed: Greg Harris , Randall Hauch --- .../apache/kafka/connect/runtime/Worker.java | 2 +- .../connect/runtime/WorkerSourceTask.java | 38 ++++++++++++---- .../kafka/connect/runtime/WorkerTask.java | 4 ++ .../ConnectWorkerIntegrationTest.java | 45 +++++++++++++++++-- .../connect/integration/ConnectorHandle.java | 8 ++-- .../MonitorableSourceConnector.java | 1 + .../runtime/ErrorHandlingTaskTest.java | 3 +- ...rrorHandlingTaskWithTopicCreationTest.java | 3 +- .../connect/runtime/WorkerSourceTaskTest.java | 5 ++- ...WorkerSourceTaskWithTopicCreationTest.java | 5 ++- .../kafka/connect/runtime/WorkerTest.java | 4 +- .../runtime/WorkerWithTopicCreationTest.java | 4 +- 12 files changed, 98 insertions(+), 24 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java index 45a93dfc4650b..fffe71a9db451 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java @@ -626,7 +626,7 @@ private WorkerTask buildWorkerTask(ClusterConfigState configState, // Note we pass the configState as it performs dynamic transformations under the covers return new WorkerSourceTask(id, (SourceTask) task, statusListener, initialState, keyConverter, valueConverter, headerConverter, transformationChain, producer, admin, topicCreationGroups, - offsetReader, offsetWriter, config, configState, metrics, loader, time, retryWithToleranceOperator, herder.statusBackingStore()); + offsetReader, offsetWriter, config, configState, metrics, loader, time, retryWithToleranceOperator, herder.statusBackingStore(), executor); } else if (task instanceof SinkTask) { TransformationChain transformationChain = new TransformationChain<>(connConfig.transformations(), retryWithToleranceOperator); log.info("Initializing: {}", transformationChain); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java index 342fa73d8724a..48660f3b33233 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java @@ -60,6 +60,7 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -86,6 +87,7 @@ class WorkerSourceTask extends WorkerTask { private final TopicAdmin admin; private final CloseableOffsetStorageReader offsetReader; private final OffsetStorageWriter offsetWriter; + private final Executor closeExecutor; private final SourceTaskMetricsGroup sourceTaskMetricsGroup; private final AtomicReference producerSendException; private final boolean isTopicTrackingEnabled; @@ -123,7 +125,8 @@ public WorkerSourceTask(ConnectorTaskId id, ClassLoader loader, Time time, RetryWithToleranceOperator retryWithToleranceOperator, - StatusBackingStore statusBackingStore) { + StatusBackingStore statusBackingStore, + Executor closeExecutor) { super(id, statusListener, initialState, loader, connectMetrics, retryWithToleranceOperator, time, statusBackingStore); @@ -139,6 +142,7 @@ public WorkerSourceTask(ConnectorTaskId id, this.admin = admin; this.offsetReader = offsetReader; this.offsetWriter = offsetWriter; + this.closeExecutor = closeExecutor; this.toSend = null; this.lastSendFailed = false; @@ -171,13 +175,9 @@ protected void close() { log.warn("Could not stop task", t); } } - if (producer != null) { - try { - producer.close(Duration.ofSeconds(30)); - } catch (Throwable t) { - log.warn("Could not close producer", t); - } - } + + closeProducer(Duration.ofSeconds(30)); + if (admin != null) { try { admin.close(Duration.ofSeconds(30)); @@ -202,6 +202,14 @@ public void removeMetrics() { public void cancel() { super.cancel(); offsetReader.close(); + // We proactively close the producer here as the main work thread for the task may + // be blocked indefinitely in a call to Producer::send if automatic topic creation is + // not enabled on either the connector or the Kafka cluster. Closing the producer should + // unblock it in that case and allow shutdown to proceed normally. + // With a duration of 0, the producer's own shutdown logic should be fairly quick, + // but closing user-pluggable classes like interceptors may lag indefinitely. So, we + // call close on a separate thread in order to avoid blocking the herder's tick thread. + closeExecutor.execute(() -> closeProducer(Duration.ZERO)); } @Override @@ -259,6 +267,16 @@ public void execute() { } } + private void closeProducer(Duration duration) { + if (producer != null) { + try { + producer.close(duration); + } catch (Throwable t) { + log.warn("Could not close producer for {}", id, t); + } + } + } + private void maybeThrowProducerSendException() { if (producerSendException.get() != null) { throw new ConnectException( @@ -488,7 +506,9 @@ public boolean commitOffsets() { while (!outstandingMessages.isEmpty()) { try { long timeoutMs = timeout - time.milliseconds(); - if (timeoutMs <= 0) { + // If the task has been cancelled, no more records will be sent from the producer; in that case, if any outstanding messages remain, + // we can stop flushing immediately + if (isCancelled() || timeoutMs <= 0) { log.error("{} Failed to flush, timed out while waiting for producer to flush outstanding {} messages", this, outstandingMessages.size()); finishFailedFlush(); recordCommitFailure(time.milliseconds() - started, null); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java index 5cd992052ba01..a717af2343220 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java @@ -156,6 +156,10 @@ protected boolean isStopping() { return stopping; } + protected boolean isCancelled() { + return cancelled; + } + private void doClose() { try { close(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java index bd981832e1174..5cd794e7c83b1 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java @@ -65,13 +65,14 @@ public class ConnectWorkerIntegrationTest { private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(30); private static final int NUM_WORKERS = 3; private static final int NUM_TASKS = 4; + private static final int MESSAGES_PER_POLL = 10; private static final String CONNECTOR_NAME = "simple-source"; private static final String TOPIC_NAME = "test-topic"; private EmbeddedConnectCluster.Builder connectBuilder; private EmbeddedConnectCluster connect; - Map workerProps = new HashMap<>(); - Properties brokerProps = new Properties(); + private Map workerProps; + private Properties brokerProps; @Rule public TestRule watcher = ConnectIntegrationTestUtils.newTestWatcher(log); @@ -79,10 +80,12 @@ public class ConnectWorkerIntegrationTest { @Before public void setup() { // setup Connect worker properties + workerProps = new HashMap<>(); workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS)); workerProps.put(CONNECTOR_CLIENT_POLICY_CLASS_CONFIG, "All"); // setup Kafka broker properties + brokerProps = new Properties(); brokerProps.put("auto.create.topics.enable", String.valueOf(false)); // build a Connect cluster backed by Kafka and Zk @@ -288,14 +291,48 @@ public void testTaskStatuses() throws Exception { decreasedNumTasks, "Connector task statuses did not update in time."); } + @Test + public void testSourceTaskNotBlockedOnShutdownWithNonExistentTopic() throws Exception { + // When automatic topic creation is disabled on the broker + brokerProps.put("auto.create.topics.enable", "false"); + connect = connectBuilder + .brokerProps(brokerProps) + .numWorkers(1) + .numBrokers(1) + .build(); + connect.start(); + + connect.assertions().assertAtLeastNumWorkersAreUp(1, "Initial group of workers did not start in time."); + + // and when the connector is not configured to create topics + Map props = defaultSourceConnectorProps("nonexistenttopic"); + props.remove(DEFAULT_TOPIC_CREATION_PREFIX + REPLICATION_FACTOR_CONFIG); + props.remove(DEFAULT_TOPIC_CREATION_PREFIX + PARTITIONS_CONFIG); + props.put("throughput", "-1"); + + ConnectorHandle connector = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME); + connector.expectedRecords(NUM_TASKS * MESSAGES_PER_POLL); + connect.configureConnector(CONNECTOR_NAME, props); + connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(CONNECTOR_NAME, + NUM_TASKS, "Connector tasks did not start in time"); + connector.awaitRecords(TimeUnit.MINUTES.toMillis(1)); + + // Then if we delete the connector, it and each of its tasks should be stopped by the framework + // even though the producer is blocked because there is no topic + StartAndStopLatch stopCounter = connector.expectedStops(1); + connect.deleteConnector(CONNECTOR_NAME); + + assertTrue("Connector and all tasks were not stopped in time", stopCounter.await(1, TimeUnit.MINUTES)); + } + private Map defaultSourceConnectorProps(String topic) { // setup up props for the source connector Map props = new HashMap<>(); props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getSimpleName()); props.put(TASKS_MAX_CONFIG, String.valueOf(NUM_TASKS)); props.put(TOPIC_CONFIG, topic); - props.put("throughput", String.valueOf(10)); - props.put("messages.per.poll", String.valueOf(10)); + props.put("throughput", "10"); + props.put("messages.per.poll", String.valueOf(MESSAGES_PER_POLL)); props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName()); props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName()); props.put(DEFAULT_TOPIC_CREATION_PREFIX + REPLICATION_FACTOR_CONFIG, String.valueOf(1)); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorHandle.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorHandle.java index 06bc37352e455..ffc9e7a7a4fb0 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorHandle.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorHandle.java @@ -292,8 +292,8 @@ public StartAndStopLatch expectedStarts(int expectedStarts, boolean includeTasks * {@link StartAndStopLatch#await(long, TimeUnit)} to wait up to a specified duration for the * connector and all tasks to be started at least the specified number of times. * - *

This method does not track the number of times the connector and tasks are stopped, and - * only tracks the number of times the connector and tasks are started. + *

This method does not track the number of times the connector and tasks are started, and + * only tracks the number of times the connector and tasks are stopped. * * @param expectedStops the minimum number of starts that are expected once this method is * called @@ -315,8 +315,8 @@ public StartAndStopLatch expectedStops(int expectedStops) { * {@link StartAndStopLatch#await(long, TimeUnit)} to wait up to a specified duration for the * connector and all tasks to be started at least the specified number of times. * - *

This method does not track the number of times the connector and tasks are stopped, and - * only tracks the number of times the connector and tasks are started. + *

This method does not track the number of times the connector and tasks are started, and + * only tracks the number of times the connector and tasks are stopped. * * @param expectedStops the minimum number of starts that are expected once this method is * called diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java index c11b9bd79d0fb..aaada374cb2b6 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java @@ -130,6 +130,7 @@ public List poll() { throttler.throttle(); } taskHandle.record(batchSize); + log.info("Returning batch of {} records", batchSize); return LongStream.range(0, batchSize) .mapToObj(i -> new SourceRecord( Collections.singletonMap("task.id", taskId), diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java index 70bbfc6590cd1..daa9eddac9ca9 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java @@ -77,6 +77,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.Executor; import static java.util.Collections.emptyMap; import static java.util.Collections.singletonList; @@ -563,7 +564,7 @@ private void createSourceTask(TargetState initialState, RetryWithToleranceOperat producer, admin, null, offsetReader, offsetWriter, workerConfig, ClusterConfigState.EMPTY, metrics, pluginLoader, time, retryWithToleranceOperator, - statusBackingStore); + statusBackingStore, (Executor) Runnable::run); } private ConsumerRecords records(ConsumerRecord record) { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskWithTopicCreationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskWithTopicCreationTest.java index 692f4d24b18b1..5cdfab96d22ee 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskWithTopicCreationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskWithTopicCreationTest.java @@ -80,6 +80,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; +import java.util.concurrent.Executor; import static java.util.Collections.emptyMap; import static java.util.Collections.singletonList; @@ -581,7 +582,7 @@ private void createSourceTask(TargetState initialState, RetryWithToleranceOperat producer, admin, TopicCreationGroup.configuredGroups(sourceConfig), offsetReader, offsetWriter, workerConfig, ClusterConfigState.EMPTY, metrics, pluginLoader, time, retryWithToleranceOperator, - statusBackingStore); + statusBackingStore, (Executor) Runnable::run); } private ConsumerRecords records(ConsumerRecord record) { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java index 8c0988735369a..0f9e6562fb99f 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java @@ -197,7 +197,7 @@ private void createWorkerTask(TargetState initialState, Converter keyConverter, workerTask = new WorkerSourceTask(taskId, sourceTask, statusListener, initialState, keyConverter, valueConverter, headerConverter, transformationChain, producer, admin, null, offsetReader, offsetWriter, config, clusterConfigState, metrics, plugins.delegatingLoader(), Time.SYSTEM, - RetryWithToleranceOperatorTest.NOOP_OPERATOR, statusBackingStore); + RetryWithToleranceOperatorTest.NOOP_OPERATOR, statusBackingStore, Runnable::run); } @Test @@ -710,6 +710,9 @@ public void testCancel() { offsetReader.close(); PowerMock.expectLastCall(); + producer.close(Duration.ZERO); + PowerMock.expectLastCall(); + PowerMock.replayAll(); workerTask.cancel(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskWithTopicCreationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskWithTopicCreationTest.java index c8fdfa72bfe98..d26faa40707dd 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskWithTopicCreationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskWithTopicCreationTest.java @@ -222,7 +222,7 @@ private void createWorkerTask(TargetState initialState, Converter keyConverter, workerTask = new WorkerSourceTask(taskId, sourceTask, statusListener, initialState, keyConverter, valueConverter, headerConverter, transformationChain, producer, admin, TopicCreationGroup.configuredGroups(sourceConfig), offsetReader, offsetWriter, config, clusterConfigState, metrics, plugins.delegatingLoader(), Time.SYSTEM, - RetryWithToleranceOperatorTest.NOOP_OPERATOR, statusBackingStore); + RetryWithToleranceOperatorTest.NOOP_OPERATOR, statusBackingStore, Runnable::run); } @Test @@ -754,6 +754,9 @@ public void testCancel() { offsetReader.close(); PowerMock.expectLastCall(); + producer.close(Duration.ZERO); + PowerMock.expectLastCall(); + PowerMock.replayAll(); workerTask.cancel(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java index cbf0802cd4b28..f3c9167ae4d7d 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java @@ -94,6 +94,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; @@ -1482,7 +1483,8 @@ private void expectNewWorkerTask() throws Exception { EasyMock.eq(pluginLoader), anyObject(Time.class), anyObject(RetryWithToleranceOperator.class), - anyObject(StatusBackingStore.class)) + anyObject(StatusBackingStore.class), + anyObject(Executor.class)) .andReturn(workerTask); } /* Name here needs to be unique as we are testing the aliasing mechanism */ diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerWithTopicCreationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerWithTopicCreationTest.java index 3bf61088ed80b..f0b959efb2ff5 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerWithTopicCreationTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerWithTopicCreationTest.java @@ -85,6 +85,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; @@ -1390,7 +1391,8 @@ private void expectNewWorkerTask() throws Exception { EasyMock.eq(pluginLoader), anyObject(Time.class), anyObject(RetryWithToleranceOperator.class), - anyObject(StatusBackingStore.class)) + anyObject(StatusBackingStore.class), + anyObject(Executor.class)) .andReturn(workerTask); }