diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 76690bbc829c3..9d300b4e656eb 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -136,7 +136,7 @@ files="(KafkaConfigBackingStore|Values).java"/> + files="(DistributedHerder|RestClient|RestServer|JsonConverter|KafkaConfigBackingStore|FileStreamSourceTask|TopicAdmin|WorkerSourceTask).java"/> , ProducerRecord> outstandingMessages; // A second buffer is used while an offset flush is running private IdentityHashMap, ProducerRecord> outstandingMessagesBacklog; - private boolean flushing; - private CountDownLatch stopRequestedLatch; + private boolean recordFlushPending; + private boolean offsetFlushPending; + private final CountDownLatch stopRequestedLatch; private Map taskConfig; private boolean started = false; @@ -144,7 +145,7 @@ public WorkerSourceTask(ConnectorTaskId id, this.lastSendFailed = false; this.outstandingMessages = new IdentityHashMap<>(); this.outstandingMessagesBacklog = new IdentityHashMap<>(); - this.flushing = false; + this.recordFlushPending = false; this.stopRequestedLatch = new CountDownLatch(1); this.sourceTaskMetricsGroup = new SourceTaskMetricsGroup(id, connectMetrics); this.producerSendException = new AtomicReference<>(); @@ -335,7 +336,7 @@ private boolean sendRecords() { // messages and update the offsets. synchronized (this) { if (!lastSendFailed) { - if (!flushing) { + if (!recordFlushPending) { outstandingMessages.put(producerRecord, producerRecord); } else { outstandingMessagesBacklog.put(producerRecord, producerRecord); @@ -453,12 +454,12 @@ private void commitTaskRecord(SourceRecord record, RecordMetadata metadata) { private synchronized void recordSent(final ProducerRecord record) { ProducerRecord removed = outstandingMessages.remove(record); // While flushing, we may also see callbacks for items in the backlog - if (removed == null && flushing) + if (removed == null && recordFlushPending) removed = outstandingMessagesBacklog.remove(record); // But if neither one had it, something is very wrong if (removed == null) { log.error("{} CRITICAL Saw callback for record that was not present in the outstanding message set: {}", this, record); - } else if (flushing && outstandingMessages.isEmpty()) { + } else if (recordFlushPending && outstandingMessages.isEmpty()) { // flush thread may be waiting on the outstanding messages to clear this.notifyAll(); } @@ -475,11 +476,15 @@ public boolean commitOffsets() { synchronized (this) { // First we need to make sure we snapshot everything in exactly the current state. This // means both the current set of messages we're still waiting to finish, stored in this - // class, which setting flushing = true will handle by storing any new values into a new + // class, which setting recordFlushPending = true will handle by storing any new values into a new // buffer; and the current set of user-specified offsets, stored in the // OffsetStorageWriter, for which we can use beginFlush() to initiate the snapshot. - flushing = true; - boolean flushStarted = offsetWriter.beginFlush(); + // No need to begin a new offset flush if we timed out waiting for records to be flushed to + // Kafka in a prior attempt. + if (!recordFlushPending) { + recordFlushPending = true; + offsetFlushPending = offsetWriter.beginFlush(); + } // Still wait for any producer records to flush, even if there aren't any offsets to write // to persistent storage @@ -490,7 +495,6 @@ public boolean commitOffsets() { long timeoutMs = timeout - time.milliseconds(); if (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); return false; } @@ -506,7 +510,7 @@ public boolean commitOffsets() { } } - if (!flushStarted) { + if (!offsetFlushPending) { // There was nothing in the offsets to process, but we still waited for the data in the // buffer to flush. This is useful since this can feed into metrics to monitor, e.g. // flush time, which can be used for monitoring even if the connector doesn't record any @@ -583,7 +587,8 @@ private synchronized void finishFailedFlush() { offsetWriter.cancelFlush(); outstandingMessages.putAll(outstandingMessagesBacklog); outstandingMessagesBacklog.clear(); - flushing = false; + recordFlushPending = false; + offsetFlushPending = false; } private synchronized void finishSuccessfulFlush() { @@ -591,7 +596,8 @@ private synchronized void finishSuccessfulFlush() { IdentityHashMap, ProducerRecord> temp = outstandingMessages; outstandingMessages = outstandingMessagesBacklog; outstandingMessagesBacklog = temp; - flushing = false; + recordFlushPending = false; + offsetFlushPending = false; } @Override 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..12e9a264b41f4 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 @@ -841,6 +841,135 @@ public void testHeadersWithCustomConverter() throws Exception { PowerMock.verifyAll(); } + @Test + public void testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstandingWithOffsets() throws Exception { + testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstanding(true); + } + + @Test + public void testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstandingWithoutOffsets() throws Exception { + testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstanding(false); + } + + private void testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstanding(boolean withOffsets) throws Exception { + createWorkerTask(); + + EasyMock.makeThreadSafe(sourceTask, false); + + sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class)); + EasyMock.expectLastCall(); + sourceTask.start(TASK_PROPS); + EasyMock.expectLastCall(); + statusListener.onStartup(taskId); + EasyMock.expectLastCall(); + + sourceTask.commit(); + EasyMock.expectLastCall().anyTimes(); + sourceTask.commitRecord(EasyMock.anyObject(), EasyMock.anyObject()); + EasyMock.expectLastCall().anyTimes(); + expectConvertHeadersAndKeyValue(true); + expectApplyTransformationChain(true); + expectTaskGetTopic(true); + offsetWriter.offset(PARTITION, OFFSET); + PowerMock.expectLastCall().anyTimes(); + EasyMock.expect(offsetWriter.beginFlush()).andReturn(withOffsets).anyTimes(); + if (withOffsets) { + Future offsetFuture = EasyMock.niceMock(Future.class); + EasyMock.expect(offsetWriter.doFlush(EasyMock.anyObject())).andReturn(offsetFuture).anyTimes(); + EasyMock.replay(offsetFuture); + } + + // We'll wait for some data, then trigger a flush + EasyMock.expect(sourceTask.poll()).andAnswer(() -> { + Thread.sleep(10); + return RECORDS; + }); + + expectTopicCreation(TOPIC); + + final CountDownLatch initialSendLatch = new CountDownLatch(1); + final Capture initialSendCallback = EasyMock.newCapture(); + EasyMock.expect(producer.send(EasyMock.anyObject(), EasyMock.capture(initialSendCallback))).andAnswer(() -> { + initialSendLatch.countDown(); + return null; + }); + + final CountDownLatch blockedPoll = new CountDownLatch(1); + EasyMock.expect(sourceTask.poll()).andAnswer(() -> { + // Block in poll here until we're ready to return the next record in the next call + blockedPoll.await(); + return null; + }); + + // Wait for another record before flushing again + EasyMock.expect(sourceTask.poll()).andAnswer(() -> { + Thread.sleep(10); + return RECORDS; + }); + final CountDownLatch subsequentSendLatch = new CountDownLatch(1); + final Capture subsequentSendCallback = EasyMock.newCapture(); + EasyMock.expect(producer.send(EasyMock.anyObject(), EasyMock.capture(subsequentSendCallback))).andAnswer(() -> { + subsequentSendLatch.countDown(); + return null; + }); + + final CountDownLatch finalPoll = new CountDownLatch(1); + EasyMock.expect(sourceTask.poll()).andAnswer(() -> { + finalPoll.await(); + return null; + }); + + sourceTask.stop(); + EasyMock.expectLastCall(); + + statusListener.onShutdown(taskId); + EasyMock.expectLastCall(); + + expectClose(); + + PowerMock.replayAll(); + + + workerTask.initialize(TASK_CONFIG); + Future taskFuture = executor.submit(workerTask); + + assertTrue(awaitLatch(initialSendLatch)); + // First offset commit fails as first record has not yet been ack'd + assertFalse(workerTask.commitOffsets()); + + // Unblock task and let it return the next record from poll + blockedPoll.countDown(); + assertTrue(awaitLatch(subsequentSendLatch)); + + // Next offset commit attempt should still fail as first record has not yet been ack'd + assertFalse(workerTask.commitOffsets()); + + // Ack the first record + initialSendCallback.getValue().onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0L, 0, 0), null); + + // Next offset commit attempt should succeed now as the first record has been ack'd + assertTrue(workerTask.commitOffsets()); + + // After a successful offset commit, the task moves onto the next batch of records, which consists solely of the second record. + // Since that record hasn't been ack'd yet, this attempt should fail + assertFalse(workerTask.commitOffsets()); + + // Ack the second record + subsequentSendCallback.getValue().onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0L, 0, 0), null); + + // Next offset commit should succeed as the record has been ack'd + assertTrue(workerTask.commitOffsets()); + + workerTask.stop(); + finalPoll.countDown(); + assertTrue(workerTask.awaitStop(6000)); + + taskFuture.get(); + assertPollMetrics(2, 2); + + PowerMock.verifyAll(); + } + private CountDownLatch expectEmptyPolls(int minimum, final AtomicInteger count) throws InterruptedException { final CountDownLatch latch = new CountDownLatch(minimum); // Note that we stub these to allow any number of calls because the thread will continue to @@ -1068,6 +1197,10 @@ private void expectOffsetFlush(boolean succeed) throws Exception { } private void assertPollMetrics(int minimumPollCountExpected) { + assertPollMetrics(minimumPollCountExpected, RECORDS.size()); + } + + private void assertPollMetrics(int minimumPollCountExpected, int activeCountMaxExpected) { MetricGroup sourceTaskGroup = workerTask.sourceTaskMetricsGroup().metricGroup(); MetricGroup taskGroup = workerTask.taskMetricsGroup().metricGroup(); double pollRate = metrics.currentMetricValueAsDouble(sourceTaskGroup, "source-record-poll-rate"); @@ -1100,7 +1233,7 @@ private void assertPollMetrics(int minimumPollCountExpected) { double activeCountMax = metrics.currentMetricValueAsDouble(sourceTaskGroup, "source-record-active-count-max"); assertEquals(0, activeCount, 0.000001d); if (minimumPollCountExpected > 0) { - assertEquals(RECORDS.size(), activeCountMax, 0.000001d); + assertEquals(activeCountMaxExpected, activeCountMax, 0.000001d); } }