diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java index ce6a60bac2ab4..d9367800b4222 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java @@ -58,7 +58,8 @@ public class MockConsumer implements Consumer { private final Set paused; private Map>> records; - private KafkaException exception; + private KafkaException pollException; + private KafkaException offsetsException; private AtomicBoolean wakeup; private boolean closed; @@ -71,7 +72,7 @@ public MockConsumer(OffsetResetStrategy offsetResetStrategy) { this.beginningOffsets = new HashMap<>(); this.endOffsets = new HashMap<>(); this.pollTasks = new LinkedList<>(); - this.exception = null; + this.pollException = null; this.wakeup = new AtomicBoolean(false); this.committed = new HashMap<>(); } @@ -170,9 +171,9 @@ public synchronized ConsumerRecords poll(final Duration timeout) { throw new WakeupException(); } - if (exception != null) { - RuntimeException exception = this.exception; - this.exception = null; + if (pollException != null) { + RuntimeException exception = this.pollException; + this.pollException = null; throw exception; } @@ -213,8 +214,20 @@ public synchronized void addRecord(ConsumerRecord record) { recs.add(record); } + /** + * @deprecated Use {@link #setPollException(KafkaException)} instead + */ + @Deprecated public synchronized void setException(KafkaException exception) { - this.exception = exception; + setPollException(exception); + } + + public synchronized void setPollException(KafkaException exception) { + this.pollException = exception; + } + + public synchronized void setOffsetsException(KafkaException exception) { + this.offsetsException = exception; } @Override @@ -388,6 +401,11 @@ public synchronized Map offsetsForTimes(Map< @Override public synchronized Map beginningOffsets(Collection partitions) { + if (offsetsException != null) { + RuntimeException exception = this.offsetsException; + this.offsetsException = null; + throw exception; + } Map result = new HashMap<>(); for (TopicPartition tp : partitions) { Long beginningOffset = beginningOffsets.get(tp); @@ -400,6 +418,11 @@ public synchronized Map beginningOffsets(Collection endOffsets(Collection partitions) { + if (offsetsException != null) { + RuntimeException exception = this.offsetsException; + this.offsetsException = null; + throw exception; + } Map result = new HashMap<>(); for (TopicPartition tp : partitions) { Long endOffset = getEndOffset(endOffsets.get(tp)); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java index 9d77d21767a89..e78276a264dc8 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java @@ -28,6 +28,7 @@ import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; @@ -312,6 +313,10 @@ public void run() { try { readToLogEnd(); log.trace("Finished read to end log for topic {}", topic); + } catch (TimeoutException e) { + log.warn("Timeout while reading log to end for topic '{}'. Retrying automatically. " + + "This may occur when brokers are unavailable or unreachable. Reason: {}", topic, e.getMessage()); + continue; } catch (WakeupException e) { // Either received another get() call and need to retry reading to end of log or stop() was // called. Both are handled by restarting this loop. diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/KafkaBasedLogTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/KafkaBasedLogTest.java index 1af6e343521a6..080f9434d13f6 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/KafkaBasedLogTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/KafkaBasedLogTest.java @@ -30,6 +30,7 @@ import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.LeaderNotAvailableException; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.record.TimestampType; @@ -370,7 +371,7 @@ public void run() { } @Test - public void testConsumerError() throws Exception { + public void testPollConsumerError() throws Exception { expectStart(); expectStop(); @@ -388,7 +389,7 @@ public void run() { consumer.schedulePollTask(new Runnable() { @Override public void run() { - consumer.setException(Errors.COORDINATOR_NOT_AVAILABLE.exception()); + consumer.setPollException(Errors.COORDINATOR_NOT_AVAILABLE.exception()); } }); @@ -423,6 +424,77 @@ public void run() { PowerMock.verifyAll(); } + @Test + public void testGetOffsetsConsumerErrorOnReadToEnd() throws Exception { + expectStart(); + + // Producer flushes when read to log end is called + producer.flush(); + PowerMock.expectLastCall(); + + expectStop(); + + PowerMock.replayAll(); + final CountDownLatch finishedLatch = new CountDownLatch(1); + Map endOffsets = new HashMap<>(); + endOffsets.put(TP0, 0L); + endOffsets.put(TP1, 0L); + consumer.updateEndOffsets(endOffsets); + store.start(); + final AtomicBoolean getInvoked = new AtomicBoolean(false); + final FutureCallback readEndFutureCallback = new FutureCallback<>(new Callback() { + @Override + public void onCompletion(Throwable error, Void result) { + getInvoked.set(true); + } + }); + consumer.schedulePollTask(new Runnable() { + @Override + public void run() { + // Once we're synchronized in a poll, start the read to end and schedule the exact set of poll events + // that should follow. This readToEnd call will immediately wakeup this consumer.poll() call without + // returning any data. + Map newEndOffsets = new HashMap<>(); + newEndOffsets.put(TP0, 1L); + newEndOffsets.put(TP1, 1L); + consumer.updateEndOffsets(newEndOffsets); + // Set exception to occur when getting offsets to read log to end. It'll be caught in the work thread, + // which will retry and eventually get the correct offsets and read log to end. + consumer.setOffsetsException(new TimeoutException("Failed to get offsets by times")); + store.readToEnd(readEndFutureCallback); + + // Should keep polling until it reaches current log end offset for all partitions + consumer.scheduleNopPollTask(); + consumer.scheduleNopPollTask(); + consumer.schedulePollTask(new Runnable() { + @Override + public void run() { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TP0_KEY, TP0_VALUE)); + consumer.addRecord(new ConsumerRecord<>(TOPIC, 1, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TP0_KEY, TP0_VALUE_NEW)); + } + }); + + consumer.schedulePollTask(new Runnable() { + @Override + public void run() { + finishedLatch.countDown(); + } + }); + } + }); + readEndFutureCallback.get(10000, TimeUnit.MILLISECONDS); + assertTrue(getInvoked.get()); + assertTrue(finishedLatch.await(10000, TimeUnit.MILLISECONDS)); + assertEquals(CONSUMER_ASSIGNMENT, consumer.assignment()); + assertEquals(1L, consumer.position(TP0)); + + store.stop(); + + assertFalse(Whitebox.getInternalState(store, "thread").isAlive()); + assertTrue(consumer.closed()); + PowerMock.verifyAll(); + } + @Test public void testProducerError() throws Exception { expectStart(); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java index 809a0b679d331..2d29916ac1b57 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java @@ -290,7 +290,7 @@ public void shouldRestoreRecordsUpToHighwatermark() { @Test public void shouldRecoverFromInvalidOffsetExceptionAndRestoreRecords() { initializeConsumer(2, 0, t1); - consumer.setException(new InvalidOffsetException("Try Again!") { + consumer.setPollException(new InvalidOffsetException("Try Again!") { public Set partitions() { return Collections.singleton(t1); } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java index c0e0de314964f..b67b664b621bf 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java @@ -236,7 +236,7 @@ public void shouldDieOnInvalidOffsetException() throws Exception { 10 * 1000, "Input record never consumed"); - mockConsumer.setException(new InvalidOffsetException("Try Again!") { + mockConsumer.setPollException(new InvalidOffsetException("Try Again!") { @Override public Set partitions() { return Collections.singleton(topicPartition); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java index b13ed535d625f..bf59c21f12c15 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java @@ -158,7 +158,7 @@ public void shouldRestoreAllMessagesFromBeginningWhenCheckpointNull() { public void shouldRecoverFromInvalidOffsetExceptionAndFinishRestore() { final int messages = 10; setupConsumer(messages, topicPartition); - consumer.setException(new InvalidOffsetException("Try Again!") { + consumer.setPollException(new InvalidOffsetException("Try Again!") { @Override public Set partitions() { return Collections.singleton(topicPartition); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java index 11485e4e0c9d0..1a7c81e72b24d 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java @@ -1234,7 +1234,7 @@ public void shouldRecoverFromInvalidOffsetExceptionOnRestoreAndFinishRestore() t () -> mockRestoreConsumer.position(changelogPartition) == 1L, "Never restore first record"); - mockRestoreConsumer.setException(new InvalidOffsetException("Try Again!") { + mockRestoreConsumer.setPollException(new InvalidOffsetException("Try Again!") { @Override public Set partitions() { return changelogPartitionSet;