From b047ca3e089662d86b77cd7fea749d7281b6110a Mon Sep 17 00:00:00 2001 From: Will James Date: Mon, 14 Oct 2019 14:59:31 -0500 Subject: [PATCH 1/3] KAKFA-8950: Fix KafkaConsumer Fetcher breaking on concurrent disconnect --- .../clients/consumer/internals/Fetcher.java | 6 ++-- .../org/apache/kafka/clients/MockClient.java | 8 ++++++ .../consumer/internals/FetcherTest.java | 28 +++++++++++++++++++ 3 files changed, 39 insertions(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 6dd76048c72b3..53d0e15a4bda2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -258,8 +258,9 @@ public synchronized int sendFetches() { if (log.isDebugEnabled()) { log.debug("Sending {} {} to broker {}", isolationLevel, data.toString(), fetchTarget); } - client.send(fetchTarget, request) - .addListener(new RequestFutureListener() { + RequestFuture future = client.send(fetchTarget, request); + this.nodesWithPendingFetchRequests.add(entry.getKey().id()); + future.addListener(new RequestFutureListener() { @Override public void onSuccess(ClientResponse resp) { synchronized (Fetcher.this) { @@ -333,7 +334,6 @@ public void onFailure(RuntimeException e) { } }); - this.nodesWithPendingFetchRequests.add(entry.getKey().id()); } return fetchRequestMap.size(); } diff --git a/clients/src/test/java/org/apache/kafka/clients/MockClient.java b/clients/src/test/java/org/apache/kafka/clients/MockClient.java index 650f670e545c5..f9fb96ce0c0a8 100644 --- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java +++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java @@ -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 connections = new HashMap<>(); @@ -254,6 +255,9 @@ public synchronized void wakeup() { numBlockingWakeups--; notify(); } + if (wakeupHook != null) { + wakeupHook.run(); + } } private synchronized void maybeAwaitWakeup() { @@ -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 diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index 9e281a7b6f822..d3fcbc334c77b 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -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; @@ -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() { From c33ee70e717530e2d8d695ae065150b4c1016e17 Mon Sep 17 00:00:00 2001 From: Will James Date: Mon, 14 Oct 2019 15:47:36 -0500 Subject: [PATCH 2/3] Fix indentation --- .../clients/consumer/internals/Fetcher.java | 126 +++++++++--------- 1 file changed, 63 insertions(+), 63 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 53d0e15a4bda2..1704e48c9d4f1 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -261,78 +261,78 @@ public synchronized int sendFetches() { RequestFuture future = client.send(fetchTarget, request); this.nodesWithPendingFetchRequests.add(entry.getKey().id()); future.addListener(new RequestFutureListener() { - @Override - public void onSuccess(ClientResponse resp) { - synchronized (Fetcher.this) { - try { - @SuppressWarnings("unchecked") - FetchResponse response = (FetchResponse) 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; - } + @Override + public void onSuccess(ClientResponse resp) { + synchronized (Fetcher.this) { + try { + @SuppressWarnings("unchecked") + FetchResponse response = (FetchResponse) 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 partitions = new HashSet<>(response.responseData().keySet()); - FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions); - - for (Map.Entry> 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 partitionData = entry.getValue(); - - log.debug("Fetch {} at offset {} for partition {} returned fetch data {}", - isolationLevel, fetchOffset, partition, partitionData); - - Iterator batches = partitionData.records.batches().iterator(); - short responseVersion = resp.requestHeader().apiVersion(); - - completedFetches.add(new CompletedFetch(partition, partitionData, - metricAggregator, batches, fetchOffset, responseVersion)); - } + Set partitions = new HashSet<>(response.responseData().keySet()); + FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions); + + for (Map.Entry> 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 partitionData = entry.getValue(); + + log.debug("Fetch {} at offset {} for partition {} returned fetch data {}", + isolationLevel, fetchOffset, partition, partitionData); + + Iterator 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()); } - }); + } + } + }); } return fetchRequestMap.size(); From 3bd1bd390d4b3ca9a5667ca376cd03d845fca97c Mon Sep 17 00:00:00 2001 From: Will James Date: Tue, 15 Oct 2019 11:56:51 -0500 Subject: [PATCH 3/3] Add comment explaining why we update nodesWithPendingFetchRequests before adding the listener --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 1704e48c9d4f1..384da315e70f4 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -259,6 +259,10 @@ public synchronized int sendFetches() { log.debug("Sending {} {} to broker {}", isolationLevel, data.toString(), fetchTarget); } RequestFuture 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() { @Override