Skip to content

KAFKA-7941: Catch TimeoutException in KafkaBasedLog worker thread #6283

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rhauch merged 5 commits into from
Aug 13, 2019
Merged

KAFKA-7941: Catch TimeoutException in KafkaBasedLog worker thread #6283

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
eb85986
KAFKA-7941: Catch TimeoutException in KafkaBasedLog worker thread
pgwhalen Feb 18, 2019
43c4fd9
Fix GlobalStateManagerImplTest after MockConsumer method rename
pgwhalen Feb 18, 2019
1c8285a
Code review tweaks
pgwhalen May 17, 2019
818923c
Preserve setException for compatibility
pgwhalen May 24, 2019
0451311
Call setException rather than replicate
pgwhalen May 24, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 29 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,8 @@ public class MockConsumer<K, V> implements Consumer<K, V> {
private final Set<TopicPartition> paused;

private Map<TopicPartition, List<ConsumerRecord<K, V>>> records;
private KafkaException exception;
private KafkaException pollException;
private KafkaException offsetsException;
private AtomicBoolean wakeup;
private boolean closed;

Expand All @@ -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<>();
}
Expand Down Expand Up @@ -170,9 +171,9 @@ public synchronized ConsumerRecords<K, V> 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;
}

Expand Down Expand Up @@ -213,8 +214,20 @@ public synchronized void addRecord(ConsumerRecord<K, V> record) {
recs.add(record);
}

/**
* @deprecated Use {@link #setPollException(KafkaException)} instead
*/
@Deprecated
public synchronized void setException(KafkaException exception) {
this.exception = exception;

@rhauch rhauch May 23, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@pgwhalen, after thinking about this some more, I think we should keep this older signature, deprecate it, and have it call setPollException(exception). Although MockConsumer is strictly speaking a non-public API, doing this will help avoid problems should people use the MockConsumer class in their tests. And it'd probably help to add JavaDoc that says to use setPollException(...) instead.

@pgwhalen pgwhalen May 24, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, I forgot it was a mildly public API. I added it back, should be good to merge now unless anything else stands out.

setPollException(exception);
}

public synchronized void setPollException(KafkaException exception) {
this.pollException = exception;
}

public synchronized void setOffsetsException(KafkaException exception) {
this.offsetsException = exception;
}

@Override
Expand Down Expand Up @@ -388,6 +401,11 @@ public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<

@Override
public synchronized Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions) {
if (offsetsException != null) {
RuntimeException exception = this.offsetsException;
this.offsetsException = null;
throw exception;
}
Map<TopicPartition, Long> result = new HashMap<>();
for (TopicPartition tp : partitions) {
Long beginningOffset = beginningOffsets.get(tp);
Expand All @@ -400,6 +418,11 @@ public synchronized Map<TopicPartition, Long> beginningOffsets(Collection<TopicP

@Override
public synchronized Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions) {
if (offsetsException != null) {
RuntimeException exception = this.offsetsException;
this.offsetsException = null;
throw exception;
}
Map<TopicPartition, Long> result = new HashMap<>();
for (TopicPartition tp : partitions) {
Long endOffset = getEndOffset(endOffsets.get(tp));
Expand Down
5 changes: 5 additions & 0 deletions connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -312,6 +313,10 @@ public void run() {
try {
readToLogEnd();
log.trace("Finished read to end log for topic {}", topic);
} catch (TimeoutException e) {

@wicknicks wicknicks May 24, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

there are a couple of other places where we call readToEnd() and poll() (links below). should we catch this exception in those places also?

kafka/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java

Line 332 in 176ea0d

poll(Integer.MAX_VALUE);

kafka/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java

Line 153 in 176ea0d

readToLogEnd();

@pgwhalen pgwhalen May 25, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the additional eyes @wicknicks !

The readToLogEnd() call is inside start()so my thinking was that failure would happen on startup which seems okay (if not desirable). If there were broker availability issues it seems reasonable to me for Connect not to start. If we want to try to be more robust we could come up with some retrying strategy, but I bet that would require changes in a number of other places.

The poll() method contains its own catch (KafkaException e) which is a super class of TimeoutException so we ought to be okay there:

kafka/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java

Line 267 in 176ea0d

} catch (KafkaException 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;

@qinghui-xu qinghui-xu Aug 12, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would suggest to backoff the retry, as the timeout probably means the unavailability of the resource (it might be the kafka broker itself, or the network, or something else), an immediate retry likely will not be success.

@pgwhalen pgwhalen Aug 12, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In practice, actual retries to the network will still be beholden to the request.timeout.ms and retry.backoff.ms inside the consumer, so that isn't an issue here.

@qinghui-xu qinghui-xu Aug 13, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's true, thanks for refreshing my mind. Then no need to backoff here.

} 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.
Expand Down
76 changes: 74 additions & 2 deletions connect/runtime/src/test/java/org/apache/kafka/connect/util/KafkaBasedLogTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -370,7 +371,7 @@ public void run() {
}

@Test
public void testConsumerError() throws Exception {
public void testPollConsumerError() throws Exception {
expectStart();
expectStop();

Expand All @@ -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());
}
});

Expand Down Expand Up @@ -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<TopicPartition, Long> endOffsets = new HashMap<>();
endOffsets.put(TP0, 0L);
endOffsets.put(TP1, 0L);
consumer.updateEndOffsets(endOffsets);
store.start();
final AtomicBoolean getInvoked = new AtomicBoolean(false);
final FutureCallback<Void> readEndFutureCallback = new FutureCallback<>(new Callback<Void>() {
@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<TopicPartition, Long> 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.<Thread>getInternalState(store, "thread").isAlive());
assertTrue(consumer.closed());
PowerMock.verifyAll();
}

@Test
public void testProducerError() throws Exception {
expectStart();
Expand Down
2 changes: 1 addition & 1 deletion ...rc/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TopicPartition> partitions() {
return Collections.singleton(t1);
}
Expand Down
2 changes: 1 addition & 1 deletion ...ms/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TopicPartition> partitions() {
return Collections.singleton(topicPartition);
Expand Down
2 changes: 1 addition & 1 deletion .../src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TopicPartition> partitions() {
return Collections.singleton(topicPartition);
Expand Down
2 changes: 1 addition & 1 deletion streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TopicPartition> partitions() {
return changelogPartitionSet;
Expand Down