Skip to content

KAKFA-8950: Fix KafkaConsumer Fetcher breaking on concurrent disconnect #7511

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
rajinisivaram merged 3 commits into from
Oct 17, 2019
Merged

KAKFA-8950: Fix KafkaConsumer Fetcher breaking on concurrent disconnect #7511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b047ca3
KAKFA-8950: Fix KafkaConsumer Fetcher breaking on concurrent disconnect
wiggzz Oct 14, 2019
c33ee70
Fix indentation
wiggzz Oct 14, 2019
3bd1bd3
Add comment explaining why we update nodesWithPendingFetchRequests be…
wiggzz Oct 15, 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
136 changes: 70 additions & 66 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,82 +258,86 @@ public synchronized int sendFetches() {
if (log.isDebugEnabled()) {
log.debug("Sending {} {} to broker {}", isolationLevel, data.toString(), fetchTarget);
}
client.send(fetchTarget, request)
.addListener(new RequestFutureListener<ClientResponse>() {
@Override
public void onSuccess(ClientResponse resp) {
synchronized (Fetcher.this) {
try {
@SuppressWarnings("unchecked")
FetchResponse<Records> response = (FetchResponse<Records>) resp.responseBody();
FetchSessionHandler handler = sessionHandler(fetchTarget.id());
if (handler == null) {
log.error("Unable to find FetchSessionHandler for node {}. Ignoring fetch response.",
fetchTarget.id());
return;
}
if (!handler.handleResponse(response)) {
return;
}
RequestFuture<ClientResponse> future = client.send(fetchTarget, request);
// We add the node to the set of nodes with pending fetch requests before adding the
// listener because the future may have been fulfilled on another thread (e.g. during a
// disconnection being handled by the heartbeat thread) which will mean the listener
// will be invoked synchronously.
this.nodesWithPendingFetchRequests.add(entry.getKey().id());
future.addListener(new RequestFutureListener<ClientResponse>() {

@thomaslee thomaslee Oct 14, 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.

Wondering if the subtleties of this fairly benign-looking code warrant a comment here or in the test?

@wiggzz wiggzz Oct 14, 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.

Was hoping the test would catch the issues with it - but I guess it's possible in the future some change causes the test to be less useful. Could stick a comment to the effect of "ensure nodesWithPendingFetchRequests is always present before finalizer could be called" or something.

@wiggzz wiggzz Oct 14, 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.

Was also wondering if there could ever be an exception thrown by addListener which would cause the listener to not be added or the completion handler to not be called?

@thomaslee thomaslee Oct 14, 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.

Was also wondering if there could ever be an exception thrown by addListener which would cause the listener to not be added or the completion handler to not be called?

Hm good question ... find it hard to imagine as implemented unless we end up with multiple listeners executing on the consumer thread & a listener that precedes this one throws or something along those lines. And in that scenario right now I think we'd expect the exception to bubble out of KafkaConsumer.poll(), which would at least give us a clear signal that something went terribly wrong.

@rajinisivaram rajinisivaram Oct 15, 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.

We rely on addListener always resuling in listener being invoked in various places. So the code should be ok. But a comment that explains the sequence and exception scenarios would be useful.

@wiggzz wiggzz Oct 15, 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.

Ok I've added a comment describing the rationale behind the sequence. If this looks good, what are the next steps for getting this merged?

@Override
public void onSuccess(ClientResponse resp) {
synchronized (Fetcher.this) {
try {
@SuppressWarnings("unchecked")
FetchResponse<Records> response = (FetchResponse<Records>) resp.responseBody();
FetchSessionHandler handler = sessionHandler(fetchTarget.id());
if (handler == null) {
log.error("Unable to find FetchSessionHandler for node {}. Ignoring fetch response.",
fetchTarget.id());
return;
}
if (!handler.handleResponse(response)) {
return;
}

Set<TopicPartition> partitions = new HashSet<>(response.responseData().keySet());
FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);

for (Map.Entry<TopicPartition, FetchResponse.PartitionData<Records>> entry : response.responseData().entrySet()) {
TopicPartition partition = entry.getKey();
FetchRequest.PartitionData requestData = data.sessionPartitions().get(partition);
if (requestData == null) {
String message;
if (data.metadata().isFull()) {
message = MessageFormatter.arrayFormat(
"Response for missing full request partition: partition={}; metadata={}",
new Object[]{partition, data.metadata()}).getMessage();
} else {
message = MessageFormatter.arrayFormat(
"Response for missing session request partition: partition={}; metadata={}; toSend={}; toForget={}",
new Object[]{partition, data.metadata(), data.toSend(), data.toForget()}).getMessage();
}

// Received fetch response for missing session partition
throw new IllegalStateException(message);
} else {
long fetchOffset = requestData.fetchOffset;
FetchResponse.PartitionData<Records> partitionData = entry.getValue();

log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
isolationLevel, fetchOffset, partition, partitionData);

Iterator<? extends RecordBatch> batches = partitionData.records.batches().iterator();
short responseVersion = resp.requestHeader().apiVersion();

completedFetches.add(new CompletedFetch(partition, partitionData,
metricAggregator, batches, fetchOffset, responseVersion));
}
Set<TopicPartition> partitions = new HashSet<>(response.responseData().keySet());
FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);

for (Map.Entry<TopicPartition, FetchResponse.PartitionData<Records>> entry : response.responseData().entrySet()) {
TopicPartition partition = entry.getKey();
FetchRequest.PartitionData requestData = data.sessionPartitions().get(partition);
if (requestData == null) {
String message;
if (data.metadata().isFull()) {
message = MessageFormatter.arrayFormat(
"Response for missing full request partition: partition={}; metadata={}",
new Object[]{partition, data.metadata()}).getMessage();
} else {
message = MessageFormatter.arrayFormat(
"Response for missing session request partition: partition={}; metadata={}; toSend={}; toForget={}",
new Object[]{partition, data.metadata(), data.toSend(), data.toForget()}).getMessage();
}

sensors.fetchLatency.record(resp.requestLatencyMs());
} finally {
nodesWithPendingFetchRequests.remove(fetchTarget.id());
// Received fetch response for missing session partition
throw new IllegalStateException(message);
} else {
long fetchOffset = requestData.fetchOffset;
FetchResponse.PartitionData<Records> partitionData = entry.getValue();

log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
isolationLevel, fetchOffset, partition, partitionData);

Iterator<? extends RecordBatch> batches = partitionData.records.batches().iterator();
short responseVersion = resp.requestHeader().apiVersion();

completedFetches.add(new CompletedFetch(partition, partitionData,
metricAggregator, batches, fetchOffset, responseVersion));
}
}

sensors.fetchLatency.record(resp.requestLatencyMs());
} finally {
nodesWithPendingFetchRequests.remove(fetchTarget.id());
}
}
}

@Override
public void onFailure(RuntimeException e) {
synchronized (Fetcher.this) {
try {
FetchSessionHandler handler = sessionHandler(fetchTarget.id());
if (handler != null) {
handler.handleError(e);
}
} finally {
nodesWithPendingFetchRequests.remove(fetchTarget.id());
}
@Override
public void onFailure(RuntimeException e) {
synchronized (Fetcher.this) {
try {
FetchSessionHandler handler = sessionHandler(fetchTarget.id());
if (handler != null) {
handler.handleError(e);
}
} finally {
nodesWithPendingFetchRequests.remove(fetchTarget.id());
}
});
}
}
});

this.nodesWithPendingFetchRequests.add(entry.getKey().id());

@wiggzz wiggzz Oct 14, 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.

This line is moved above to https://github.com/apache/kafka/pull/7511/files#diff-b45245913eaae46aa847d2615d62cde0R262 the rest of the changes in this file are indentation changes.

@rcrocker rcrocker Oct 21, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

(@thomaslee reminds me that send() can throw and then we're left with a dangling reference)

}
return fetchRequestMap.size();
}
Expand Down
8 changes: 8 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/MockClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@ public FutureResponse(Node node,
}

private int correlation;
private Runnable wakeupHook;
private final Time time;
private final MockMetadataUpdater metadataUpdater;
private final Map<String, ConnectionState> connections = new HashMap<>();
Expand Down Expand Up @@ -254,6 +255,9 @@ public synchronized void wakeup() {
numBlockingWakeups--;
notify();
}
if (wakeupHook != null) {
wakeupHook.run();
}
}

private synchronized void maybeAwaitWakeup() {
Expand Down Expand Up @@ -545,6 +549,10 @@ public Node leastLoadedNode(long now) {
return null;
}

public void setWakeupHook(Runnable wakeupHook) {
this.wakeupHook = wakeupHook;
}

/**
* The RequestMatcher provides a way to match a particular request to a response prepared
* through {@link #prepareResponse(RequestMatcher, AbstractResponse)}. Basically this allows testers
Expand Down
28 changes: 28 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -3772,6 +3773,33 @@ public void testPreferredReadReplicaOffsetError() {
assertEquals(selected.id(), -1);
}

@Test
public void testFetchCompletedBeforeHandlerAdded() {
buildFetcher();
assignFromUser(singleton(tp0));
subscriptions.seek(tp0, 0);
fetcher.sendFetches();
client.prepareResponse(fullFetchResponse(tp0, buildRecords(1L, 1, 1), Errors.NONE, 100L, 0));
consumerClient.poll(time.timer(0));
fetchedRecords();
Node node = fetcher.selectReadReplica(tp0, subscriptions.position(tp0).currentLeader.leader, time.milliseconds());

AtomicBoolean wokenUp = new AtomicBoolean(false);
client.setWakeupHook(() -> {
if (!wokenUp.getAndSet(true)) {
consumerClient.disconnectAsync(node);
consumerClient.poll(time.timer(0));
}
});

assertEquals(1, fetcher.sendFetches());

consumerClient.disconnectAsync(node);
consumerClient.poll(time.timer(0));

assertEquals(1, fetcher.sendFetches());
}

private MockClient.RequestMatcher listOffsetRequestMatcher(final long timestamp) {
// matches any list offset request with the provided timestamp
return new MockClient.RequestMatcher() {
Expand Down