KAFKA-12497: Skip periodic offset commits for failed source tasks

build.gradle

@@ -1274,6 +1274,7 @@ project(':clients') {
 
     testImplementation libs.bcpkix
     testImplementation libs.junitJupiter
+    testImplementation libs.log4j
     testImplementation libs.mockitoInline
 
     testRuntimeOnly libs.slf4jlog4j

checkstyle/import-control.xml

@@ -198,6 +198,7 @@
 
     <subpackage name="utils">
       <allow pkg="org.apache.kafka.common" />
+      <allow pkg="org.apache.log4j" />
     </subpackage>
 
     <subpackage name="quotas">
@@ -454,9 +455,6 @@
         <allow pkg="com.fasterxml.jackson" />
         <allow pkg="kafka.utils" />
         <allow pkg="org.apache.zookeeper" />
-        <subpackage name="testutil">
-          <allow pkg="org.apache.log4j" />
-        </subpackage>
       </subpackage>
     </subpackage>
   </subpackage>
@@ -573,7 +571,6 @@
         <allow pkg="com.fasterxml.jackson" />
         <allow pkg="org.apache.http"/>
         <allow pkg="io.swagger.v3.oas.annotations"/>
-        <allow pkg="kafka.utils" />
         <subpackage name="resources">
           <allow pkg="org.apache.log4j" />
         </subpackage>

streams/src/test/java/org/apache/kafka/streams/processor/internals/testutil/LogCaptureAppender.java → clients/src/test/java/org/apache/kafka/common/utils/LogCaptureAppender.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.kafka.streams.processor.internals.testutil;
+package org.apache.kafka.common.utils;
 
 import org.apache.log4j.AppenderSkeleton;
 import org.apache.log4j.Level;

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java

@@ -104,7 +104,13 @@ public void remove(ConnectorTaskId id) {
         }
     }
 
-    private void commit(WorkerSourceTask workerTask) {
+    // Visible for testing
+    static void commit(WorkerSourceTask workerTask) {
+        if (!workerTask.shouldCommitOffsets()) {
+            log.trace("{} Skipping offset commit as there are no offsets that should be committed", workerTask);
+            return;
+        }
+
         log.debug("{} Committing offsets", workerTask);
         try {
             if (workerTask.commitOffsets()) {

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java

@@ -194,6 +194,14 @@ protected void finalOffsetCommit(boolean failed) {
         commitOffsets();
     }
 
+    /**
+     * @return whether an attempt to commit offsets should be made for the task (i.e., there are pending uncommitted
+     * offsets and the task's producer has not already failed to send a record with a non-retriable error).
+     */
+    public boolean shouldCommitOffsets() {
+        return !isFailed();
+    }
+
     public boolean commitOffsets() {
         long commitTimeoutMs = workerConfig.getLong(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG);
 

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java

@@ -63,6 +63,7 @@ abstract class WorkerTask implements Runnable {
     private final CountDownLatch shutdownLatch = new CountDownLatch(1);
     private final TaskMetricsGroup taskMetricsGroup;
     private volatile TargetState targetState;
+    private volatile boolean failed;
     private volatile boolean stopping;   // indicates whether the Worker has asked the task to stop
     private volatile boolean cancelled;  // indicates whether the Worker has cancelled the task (e.g. because of slow shutdown)
     private final ErrorHandlingMetrics errorMetrics;
@@ -84,6 +85,7 @@ public WorkerTask(ConnectorTaskId id,
         this.statusListener = taskMetricsGroup;
         this.loader = loader;
         this.targetState = initialState;
+        this.failed = false;
         this.stopping = false;
         this.cancelled = false;
         this.taskMetricsGroup.recordState(this.targetState);
@@ -163,6 +165,10 @@ public void removeMetrics() {
 
     protected abstract void close();
 
+    protected boolean isFailed() {
+        return failed;
+    }
+
     protected boolean isStopping() {
         return stopping;
     }
@@ -196,6 +202,7 @@ private void doRun() throws InterruptedException {
             statusListener.onStartup(id);
             execute();
         } catch (Throwable t) {
+            failed = true;
             if (cancelled) {
                 log.warn("{} After being scheduled for shutdown, the orphan task threw an uncaught exception. A newer instance of this task might be already running", this, t);
             } else if (stopping) {

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java

@@ -53,6 +53,7 @@
 import org.apache.kafka.connect.util.ParameterizedTest;
 import org.apache.kafka.connect.util.TopicAdmin;
 import org.apache.kafka.connect.util.TopicCreationGroup;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.easymock.Capture;
 import org.easymock.EasyMock;
 import org.easymock.IAnswer;
@@ -99,6 +100,7 @@
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.REPLICATION_FACTOR_CONFIG;
 import static org.apache.kafka.connect.runtime.WorkerConfig.TOPIC_CREATION_ENABLE_CONFIG;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
@@ -372,7 +374,7 @@ public void testFailureInPoll() throws Exception {
 
         sourceTask.stop();
         EasyMock.expectLastCall();
-        expectOffsetFlush(true);
+        expectEmptyOffsetFlush();
 
         expectClose();
 
@@ -382,8 +384,9 @@ public void testFailureInPoll() throws Exception {
         Future<?> taskFuture = executor.submit(workerTask);
 
         assertTrue(awaitLatch(pollLatch));
-        //Failure in poll should trigger automatic stop of the worker
+        //Failure in poll should trigger automatic stop of the task
         assertTrue(workerTask.awaitStop(1000));
+        assertShouldSkipCommit();
 
         taskFuture.get();
         assertPollMetrics(0);
@@ -467,6 +470,7 @@ public void testFailureInPollAfterStop() throws Exception {
         workerTask.stop();
         workerStopLatch.countDown();
         assertTrue(workerTask.awaitStop(1000));
+        assertShouldSkipCommit();
 
         taskFuture.get();
         assertPollMetrics(0);
@@ -483,11 +487,11 @@ public void testPollReturnsNoRecords() throws Exception {
 
         // We'll wait for some data, then trigger a flush
         final CountDownLatch pollLatch = expectEmptyPolls(1, new AtomicInteger());
-        expectOffsetFlush(true);
+        expectEmptyOffsetFlush();
 
         sourceTask.stop();
         EasyMock.expectLastCall();
-        expectOffsetFlush(true);
+        expectEmptyOffsetFlush();
 
         statusListener.onShutdown(taskId);
         EasyMock.expectLastCall();
@@ -528,7 +532,7 @@ public void testCommit() throws Exception {
 
         sourceTask.stop();
         EasyMock.expectLastCall();
-        expectOffsetFlush(true);
+        expectEmptyOffsetFlush();
 
         statusListener.onShutdown(taskId);
         EasyMock.expectLastCall();
@@ -647,10 +651,6 @@ public void testSendRecordsProducerCallbackFail() throws Exception {
 
     @Test
     public void testSendRecordsProducerSendFailsImmediately() {
-        if (!enableTopicCreation)
-            // should only test with topic creation enabled
-            return;
-
         createWorkerTask();
 
         SourceRecord record1 = new SourceRecord(PARTITION, OFFSET, TOPIC, 1, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
@@ -1023,6 +1023,12 @@ private void expectOffsetFlush(boolean succeed) throws Exception {
         }
     }
 
+    private void expectEmptyOffsetFlush() throws Exception {
+        EasyMock.expect(offsetWriter.beginFlush()).andReturn(false);
+        sourceTask.commit();
+        EasyMock.expectLastCall();
+    }
+
     private void assertPollMetrics(int minimumPollCountExpected) {
         MetricGroup sourceTaskGroup = workerTask.sourceTaskMetricsGroup().metricGroup();
         MetricGroup taskGroup = workerTask.taskMetricsGroup().metricGroup();
@@ -1109,4 +1115,20 @@ private void expectTopicCreation(String topic) {
             EasyMock.expect(admin.createOrFindTopics(EasyMock.capture(newTopicCapture))).andReturn(createdTopic(topic));
         }
     }
+
+    private void assertShouldSkipCommit() {
+        assertFalse(workerTask.shouldCommitOffsets());
+
+        LogCaptureAppender.setClassLoggerToTrace(SourceTaskOffsetCommitter.class);
+        LogCaptureAppender.setClassLoggerToTrace(WorkerSourceTask.class);
+        try (LogCaptureAppender committerAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class);
+             LogCaptureAppender taskAppender = LogCaptureAppender.createAndRegister(WorkerSourceTask.class)) {
+            SourceTaskOffsetCommitter.commit(workerTask);
+            assertEquals(Collections.emptyList(), taskAppender.getMessages());
+
+            List<String> committerMessages = committerAppender.getMessages();
+            assertEquals(1, committerMessages.size());
+            assertTrue(committerMessages.get(0).contains("Skipping offset commit"));
+        }
+    }
 }

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.connect.runtime.rest;
 
-import kafka.utils.LogCaptureAppender;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.http.HttpHost;
@@ -31,6 +30,7 @@
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.connect.rest.ConnectRestExtension;
 import org.apache.kafka.connect.runtime.Herder;
 import org.apache.kafka.connect.runtime.WorkerConfig;
@@ -362,10 +362,10 @@ public void testRequestLogs() throws IOException, InterruptedException {
         // Stop the server to flush all logs
         server.stop();
 
-        Collection<String> logMessages = restServerAppender.getRenderedMessages();
+        Collection<String> logMessages = restServerAppender.getMessages();
         LogCaptureAppender.unregister(restServerAppender);
         restServerAppender.close();
-        String expectedlogContent = "\"GET / HTTP/1.1\" " + String.valueOf(response.getStatusLine().getStatusCode());
+        String expectedlogContent = "\"GET / HTTP/1.1\" " + response.getStatusLine().getStatusCode();
         assertTrue(logMessages.stream().anyMatch(logMessage -> logMessage.contains(expectedlogContent)));
     }
 

core/src/test/scala/unit/kafka/utils/LogCaptureAppender.scala

@@ -20,7 +20,6 @@ package kafka.utils
 import org.apache.log4j.{AppenderSkeleton, Level, Logger}
 import org.apache.log4j.spi.LoggingEvent
 
-import scala.jdk.CollectionConverters._
 import scala.collection.mutable.ListBuffer
 
 class LogCaptureAppender extends AppenderSkeleton {
@@ -38,11 +37,6 @@ class LogCaptureAppender extends AppenderSkeleton {
     }
   }
 
-  def getRenderedMessages: java.util.List[String] = {
-    return getMessages.map(e => e.getRenderedMessage).asJava
-  }
-
-
   override def close(): Unit = {
     events.synchronized {
       events.clear()

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

@@ -58,7 +58,7 @@
 import org.apache.kafka.streams.processor.internals.ThreadMetadataImpl;
 import org.apache.kafka.streams.processor.internals.TopologyMetadata;
 import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.Stores;

streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java

@@ -33,7 +33,7 @@
 import org.apache.kafka.streams.processor.FailOnInvalidTimestamp;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 import org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;

streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java

@@ -19,6 +19,7 @@
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
@@ -31,7 +32,6 @@
 import org.apache.kafka.streams.kstream.Transformer;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.PunctuationType;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
 import org.apache.kafka.test.TestUtils;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java

@@ -35,7 +35,7 @@
 import org.apache.kafka.streams.kstream.TimeWindows;
 import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.kstream.Windows;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.SessionStore;
 import org.apache.kafka.streams.state.ValueAndTimestamp;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java

@@ -33,7 +33,7 @@
 import org.apache.kafka.streams.kstream.StreamJoined;
 import org.apache.kafka.streams.processor.internals.InternalTopicConfig;
 import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.streams.TestInputTopic;
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.streams.state.WindowBytesStoreSupplier;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java

@@ -31,6 +31,7 @@
 import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.KeyValueTimestamp;
 import org.apache.kafka.streams.StreamsBuilder;
@@ -43,7 +44,6 @@
 import org.apache.kafka.streams.kstream.Joined;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
 import org.apache.kafka.test.MockApiProcessor;
 import org.apache.kafka.test.MockApiProcessorSupplier;
 import org.apache.kafka.test.MockValueJoiner;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java

@@ -37,8 +37,8 @@
 import org.apache.kafka.streams.processor.internals.metrics.TaskMetrics;
 import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
 import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender.Event;
+import org.apache.kafka.common.utils.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender.Event;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.SessionBytesStoreSupplier;
 import org.apache.kafka.streams.state.SessionStore;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregateTest.java

@@ -39,8 +39,8 @@
 import org.apache.kafka.streams.kstream.SlidingWindows;
 import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
 import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender.Event;
+import org.apache.kafka.common.utils.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender.Event;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.streams.state.ValueAndTimestamp;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.KeyValueTimestamp;
@@ -47,7 +48,6 @@
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.Record;
 import org.apache.kafka.streams.processor.internals.ProcessorNode;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.streams.state.TimestampedWindowStore;
 import org.apache.kafka.streams.state.WindowBytesStoreSupplier;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java

@@ -30,7 +30,7 @@
 import org.apache.kafka.streams.processor.api.MockProcessorContext;
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.Record;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.test.TestRecord;
 import org.apache.kafka.test.MockApiProcessor;

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java

@@ -34,7 +34,7 @@
 import org.apache.kafka.streams.processor.api.MockProcessorContext;
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.Record;
-import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.streams.test.TestRecord;
 import org.apache.kafka.test.MockApiProcessor;