diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java index c12246f139d76..659dadfd87348 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java @@ -32,6 +32,7 @@ import org.apache.kafka.common.metrics.stats.Rate; import org.apache.kafka.common.metrics.stats.Value; import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; import org.apache.kafka.connect.data.SchemaAndValue; import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.connect.errors.RetriableException; @@ -175,6 +176,7 @@ protected void close() { } catch (Throwable t) { log.warn("Could not close transformation chain", t); } + Utils.closeQuietly(retryWithToleranceOperator, "retry operator"); } @Override 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 da2f9ef7c9245..dc7474766a034 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 @@ -29,6 +29,7 @@ import org.apache.kafka.common.metrics.stats.Rate; import org.apache.kafka.common.metrics.stats.Value; import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.connect.errors.RetriableException; import org.apache.kafka.connect.header.Header; @@ -170,6 +171,7 @@ protected void close() { } catch (Throwable t) { log.warn("Could not close transformation chain", t); } + Utils.closeQuietly(retryWithToleranceOperator, "retry operator"); } @Override diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/DeadLetterQueueReporter.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/DeadLetterQueueReporter.java index fc6181d4904de..20ed2f2d577f2 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/DeadLetterQueueReporter.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/DeadLetterQueueReporter.java @@ -204,4 +204,9 @@ private byte[] toBytes(String value) { return null; } } + + @Override + public void close() { + kafkaProducer.close(); + } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorReporter.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorReporter.java index 58336163fbf4a..5eaa42716b6b3 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorReporter.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ErrorReporter.java @@ -19,7 +19,7 @@ /** * Report an error using the information contained in the {@link ProcessingContext}. */ -public interface ErrorReporter { +public interface ErrorReporter extends AutoCloseable { /** * Report an error. @@ -28,4 +28,6 @@ public interface ErrorReporter { */ void report(ProcessingContext context); + @Override + default void close() { } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ProcessingContext.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ProcessingContext.java index f826d74fe70cb..e7fb03185f85e 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ProcessingContext.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/ProcessingContext.java @@ -18,6 +18,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.connect.source.SourceRecord; import java.util.Collection; @@ -28,7 +29,7 @@ * Contains all the metadata related to the currently evaluating operation. Only one instance of this class is meant * to exist per task in a JVM. */ -class ProcessingContext { +class ProcessingContext implements AutoCloseable { private Collection reporters = Collections.emptyList(); @@ -216,4 +217,19 @@ public void reporters(Collection reporters) { this.reporters = reporters; } + @Override + public void close() { + ConnectException e = null; + for (ErrorReporter reporter : reporters) { + try { + reporter.close(); + } catch (Throwable t) { + e = e != null ? e : new ConnectException("Failed to close all reporters"); + e.addSuppressed(t); + } + } + if (e != null) { + throw e; + } + } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java index 2513514475582..4e627ef2e70c9 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java @@ -51,7 +51,7 @@ * then it is wrapped into a ConnectException and rethrown to the caller. *

*/ -public class RetryWithToleranceOperator { +public class RetryWithToleranceOperator implements AutoCloseable { private static final Logger log = LoggerFactory.getLogger(RetryWithToleranceOperator.class); @@ -270,4 +270,9 @@ public void consumerRecord(ConsumerRecord consumedMessage) { public boolean failed() { return this.context.failed(); } + + @Override + public void close() { + this.context.close(); + } } 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 d4a34bb19818d..f3681593ef63f 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 @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.runtime; +import java.util.Arrays; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; @@ -33,6 +34,7 @@ import org.apache.kafka.connect.json.JsonConverter; import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; import org.apache.kafka.connect.runtime.errors.ErrorHandlingMetrics; +import org.apache.kafka.connect.runtime.errors.ErrorReporter; import org.apache.kafka.connect.runtime.errors.LogReporter; import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator; import org.apache.kafka.connect.runtime.errors.ToleranceType; @@ -165,6 +167,93 @@ public void tearDown() { } } + @Test + public void testSinkTasksCloseErrorReporters() throws Exception { + ErrorReporter reporter = EasyMock.mock(ErrorReporter.class); + + RetryWithToleranceOperator retryWithToleranceOperator = operator(); + retryWithToleranceOperator.metrics(errorHandlingMetrics); + retryWithToleranceOperator.reporters(singletonList(reporter)); + + createSinkTask(initialState, retryWithToleranceOperator); + + expectInitializeTask(); + reporter.close(); + EasyMock.expectLastCall(); + sinkTask.stop(); + EasyMock.expectLastCall(); + + consumer.close(); + EasyMock.expectLastCall(); + + PowerMock.replayAll(); + + workerSinkTask.initialize(TASK_CONFIG); + workerSinkTask.initializeAndStart(); + workerSinkTask.close(); + + PowerMock.verifyAll(); + } + + @Test + public void testSourceTasksCloseErrorReporters() { + ErrorReporter reporter = EasyMock.mock(ErrorReporter.class); + + RetryWithToleranceOperator retryWithToleranceOperator = operator(); + retryWithToleranceOperator.metrics(errorHandlingMetrics); + retryWithToleranceOperator.reporters(singletonList(reporter)); + + createSourceTask(initialState, retryWithToleranceOperator); + + sourceTask.stop(); + PowerMock.expectLastCall(); + + producer.close(EasyMock.anyObject()); + PowerMock.expectLastCall(); + + reporter.close(); + EasyMock.expectLastCall(); + + PowerMock.replayAll(); + + workerSourceTask.initialize(TASK_CONFIG); + workerSourceTask.close(); + + PowerMock.verifyAll(); + } + + @Test + public void testCloseErrorReportersExceptionPropagation() { + ErrorReporter reporterA = EasyMock.mock(ErrorReporter.class); + ErrorReporter reporterB = EasyMock.mock(ErrorReporter.class); + + RetryWithToleranceOperator retryWithToleranceOperator = operator(); + retryWithToleranceOperator.metrics(errorHandlingMetrics); + retryWithToleranceOperator.reporters(Arrays.asList(reporterA, reporterB)); + + createSourceTask(initialState, retryWithToleranceOperator); + + sourceTask.stop(); + PowerMock.expectLastCall(); + + producer.close(EasyMock.anyObject()); + PowerMock.expectLastCall(); + + // Even though the reporters throw exceptions, they should both still be closed. + reporterA.close(); + EasyMock.expectLastCall().andThrow(new RuntimeException()); + + reporterB.close(); + EasyMock.expectLastCall().andThrow(new RuntimeException()); + + PowerMock.replayAll(); + + workerSourceTask.initialize(TASK_CONFIG); + workerSourceTask.close(); + + PowerMock.verifyAll(); + } + @Test public void testErrorHandlingInSinkTasks() throws Exception { Map reportProps = new HashMap<>(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/ErrorReporterTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/ErrorReporterTest.java index 00a922f76ad97..f01cd4995c052 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/ErrorReporterTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/errors/ErrorReporterTest.java @@ -146,6 +146,20 @@ public void testReportDLQTwice() { PowerMock.verifyAll(); } + @Test + public void testCloseDLQ() { + DeadLetterQueueReporter deadLetterQueueReporter = new DeadLetterQueueReporter( + producer, config(singletonMap(SinkConnectorConfig.DLQ_TOPIC_NAME_CONFIG, DLQ_TOPIC)), TASK_ID, errorHandlingMetrics); + + producer.close(); + EasyMock.expectLastCall(); + replay(producer); + + deadLetterQueueReporter.close(); + + PowerMock.verifyAll(); + } + @Test public void testLogOnDisabledLogReporter() { LogReporter logReporter = new LogReporter(TASK_ID, config(emptyMap()), errorHandlingMetrics);