This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2cc6ef65e5f KAFKA-20736: Close empty share sessions safely (#22888)
2cc6ef65e5f is described below
commit 2cc6ef65e5f34aa97ecf5d013fb7e5ff7942af07
Author: Andrew Schofield <[email protected]>
AuthorDate: Thu Jul 30 07:56:09 2026 +0100
KAFKA-20736: Close empty share sessions safely (#22888)
When all of the partitions have been removed from a share session
because the share partitions assigned to a share consumer are no longer
on the broker, the share session can be closed. It is important to wait
until the share consumer has no outstanding acknowledgements to send to
the broker (these can be completed even after a partition has been
revoked from the consumer). This PR completes the share session
improvements in AK 4.4.
Reviewers: Shivsundar R <[email protected]>, Apoorv Mittal
<[email protected]>
---
checkstyle/suppressions.xml | 2 +-
.../consumer/internals/Acknowledgements.java | 8 +-
.../consumer/internals/ShareCompletedFetch.java | 18 +
.../internals/ShareConsumeRequestManager.java | 262 +++++++-----
.../consumer/internals/ShareFetchBuffer.java | 48 ++-
.../consumer/internals/ShareSessionHandler.java | 11 +-
.../internals/ShareCompletedFetchTest.java | 72 ++++
.../internals/ShareConsumeRequestManagerTest.java | 457 +++++++++++++++++----
.../consumer/internals/ShareFetchBufferTest.java | 149 ++++++-
.../internals/ShareSessionHandlerTest.java | 40 ++
10 files changed, 871 insertions(+), 196 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 4f2c312f248..5ebf937d7c9 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -98,7 +98,7 @@
files="(OffsetFetcher|RequestResponse)Test.java"/>
<suppress checks="JavaNCSS"
-
files="RequestResponseTest.java|FetcherTest.java|FetchRequestManagerTest.java|KafkaAdminClientTest.java|ConsumerMembershipManagerTest.java"/>
+
files="RequestResponseTest.java|FetcherTest.java|FetchRequestManagerTest.java|KafkaAdminClientTest.java|ConsumerMembershipManagerTest.java|ShareConsumeRequestManagerTest.java"/>
<suppress checks="NPathComplexity"
files="MemoryRecordsTest|MetricsTest|RequestResponseTest|TestSslUtils"/>
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java
index a60d24520d0..2dfd4e3388c 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java
@@ -130,8 +130,12 @@ public class Acknowledgements {
* @param acknowledgeException the exception (will be null if successful)
*/
public void complete(KafkaException acknowledgeException) {
- this.acknowledgeException = acknowledgeException;
- completed = true;
+ if (!completed) {
+ if (acknowledgeException != null) {
+ this.acknowledgeException = acknowledgeException;
+ }
+ completed = true;
+ }
}
/**
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java
index f172949ec3a..ec0e7c7db0b 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetch.java
@@ -75,6 +75,7 @@ public class ShareCompletedFetch {
private KafkaException cachedRecordException = null;
private boolean isConsumed = false;
private boolean initialized = false;
+ private ShareInFlightBatch<?, ?> deliveredBatch;
private final List<OffsetAndDeliveryCount> acquiredRecordList;
private ListIterator<OffsetAndDeliveryCount> acquiredRecordIterator;
private OffsetAndDeliveryCount nextAcquired;
@@ -136,6 +137,22 @@ public class ShareCompletedFetch {
return isConsumed;
}
+ /**
+ * Track the batch that carries records delivered to the application so we
can tell when their acknowledgements
+ * are still outstanding. We only advance to a new batch once the previous
one has been drained of its
+ * in-flight records and its acknowledgements have been taken by the
background thread to be sent.
+ */
+ void trackDeliveredBatch(ShareInFlightBatch<?, ?> inFlightBatch) {
+ if (deliveredBatch == null || (deliveredBatch.numRecords() == 0 &&
!deliveredBatch.hasRenewals())) {
+ deliveredBatch = inFlightBatch;
+ }
+ }
+
+ public boolean hasPendingAcknowledgements() {
+ // Records being renewed are removed and readded to the in-flight set
as the renewals are confirmed.
+ return deliveredBatch != null && (deliveredBatch.numRecords() > 0 ||
deliveredBatch.hasRenewals());
+ }
+
/**
* Draining a {@link ShareCompletedFetch} will signal that the data has
been consumed and the underlying resources
* are closed. This is somewhat analogous to {@link Closeable#close()
closing}, though no error will result if a
@@ -177,6 +194,7 @@ public class ShareCompletedFetch {
final boolean checkCrcs) {
// Creating an empty ShareInFlightBatch
ShareInFlightBatch<K, V> inFlightBatch = new
ShareInFlightBatch<>(nodeId, partition, acquisitionLockTimeoutMs);
+ trackDeliveredBatch(inFlightBatch);
if (cachedBatchException != null) {
// In the event that a CRC check fails, reject the entire record
batch because it is corrupt.
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java
index d9399148784..5547531bc96 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java
@@ -40,6 +40,7 @@ import
org.apache.kafka.common.requests.ShareAcknowledgeRequest;
import org.apache.kafka.common.requests.ShareAcknowledgeResponse;
import org.apache.kafka.common.requests.ShareFetchRequest;
import org.apache.kafka.common.requests.ShareFetchResponse;
+import org.apache.kafka.common.requests.ShareRequestMetadata;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.utils.internals.BufferSupplier;
@@ -246,10 +247,17 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
}
}
+ // Calculate the set of nodes which have records buffered for the
consumer to consume and acknowledge.
+ // Even when assignments change, we do not close the share sessions
which have buffered records.
+ // This also does bookkeeping of the nodes which had pending
acknowledgements in case they have become
+ // eligible for cleaning up.
+ Set<Integer> bufferedNodes = shareFetchBuffer.bufferedNodes();
+
// Iterate over the session handlers to see if there are
acknowledgements to be sent for partitions
// which are no longer part of the current subscription or which have
disappeared from the metadata.
// Also include session handlers which have no fetchable partitions so
the share sessions can be
- // brought up to date.
+ // brought up to date. Finally, close share sessions for nodes which
no longer have any share partitions
+ // and which do not have buffered data.
sessionHandlers.forEach((nodeId, sessionHandler) -> {
Node node = cluster.nodeById(nodeId);
if (node == null) {
@@ -258,6 +266,10 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
}
if (nodesWithPendingRequests.contains(nodeId)) {
log.trace("Skipping fetch because previous request to {} has
not been processed", nodeId);
+ } else if (isSessionCloseCandidate(node, sessionHandler,
handlerMap, bufferedNodes)) {
+ log.debug("Closing share session for node {} which is now
empty", nodeId);
+ sessionHandler.notifyClose();
+ handlerMap.put(node, sessionHandler);
} else {
prepareRemainingSessionHandlerForRequest(node, sessionHandler,
handlerMap);
}
@@ -340,7 +352,8 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
Map<TopicIdPartition, Acknowledgements> nodeAcksFromFetchMap =
fetchAcknowledgementsToSend.get(node.id());
if (nodeAcksFromFetchMap != null) {
nodeAcksFromFetchMap.forEach((tip, acks) -> {
- if (!isLeaderKnownToHaveChanged(node.id(), tip)) {
+ LeaderIdAndEpoch leader = shareSessionLeaderMap.get(tip);
+ if (leader != null && leader.leaderId == node.id()) {
// Check if the share session epoch is valid for sending
acknowledgements.
if (!maybeAddAcknowledgements(sessionHandler, node, tip,
acks)) {
return;
@@ -352,7 +365,7 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
log.debug("Added fetch request for previously subscribed
partition {} to node {}", tip, node.id());
} else {
log.debug("Leader for the partition is down or has
changed, failing acknowledgements for partition {}", tip);
- acks.complete(Errors.NETWORK_EXCEPTION.exception());
+ acks.complete(acknowledgementsCannotBeSentError(node.id(),
tip).exception());
maybeSendShareAcknowledgementEvent(Map.of(tip, acks),
true, Optional.empty());
}
});
@@ -364,6 +377,49 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
}
}
+ /**
+ * A share session can be closed if it's empty and has no buffered data.
+ *
+ * @return True if the node's share session can be closed.
+ */
+ private boolean isSessionCloseCandidate(Node node,
+ ShareSessionHandler sessionHandler,
+ Map<Node, ShareSessionHandler>
handlerMap,
+ Set<Integer> bufferedNodes) {
+ final int nodeId = node.id();
+ return !closing // the close path
already managed closing share sessions
+ && sessionHandler.isSessionEmpty() // no share partitions
remain in the share session
+ && !sessionHandler.isNewSession() // the session is
established on the broker
+ && !handlerMap.containsKey(node) // the handler is
already being used in this poll
+ && !bufferedNodes.contains(nodeId) // no records for this
node remain in the fetch buffer
+ && !nodeHasPendingAcknowledgements(nodeId); // the node has
pending acknowledgements
+ }
+
+ /**
+ * Returns whether there are acknowledgements still being delivered for
the node, whether waiting to be sent,
+ * in flight, or being processed as part of an acknowledge request state.
+ */
+ private boolean nodeHasPendingAcknowledgements(int nodeId) {
+ Map<TopicIdPartition, Acknowledgements> acksToSend =
fetchAcknowledgementsToSend.get(nodeId);
+ if (acksToSend != null && !acksToSend.isEmpty()) {
+ return true;
+ }
+
+ Map<TopicIdPartition, Acknowledgements> acksInFlight =
fetchAcknowledgementsInFlight.get(nodeId);
+ if (acksInFlight != null && !acksInFlight.isEmpty()) {
+ return true;
+ }
+
+ Tuple<AcknowledgeRequestState> requestStates =
acknowledgeRequestStates.get(nodeId);
+ if (requestStates != null) {
+ return
areRequestStatesInProgress(requestStates.getSyncRequestQueue())
+ || isRequestStateInProgress(requestStates.getAsyncRequest())
+ || isRequestStateInProgress(requestStates.getCloseRequest());
+ }
+
+ return false;
+ }
+
/**
* Remove session handlers for nodes which are now missing from the
cluster metadata. Only nodes with no
* in-flight requests are removed, since those requests will complete.
Piggyback acknowledgements for the
@@ -620,35 +676,21 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
public CompletableFuture<Map<TopicIdPartition, Acknowledgements>>
commitSync(
final Map<TopicIdPartition, NodeAcknowledgements>
acknowledgementsMap,
final long deadlineMs) {
- final Cluster cluster = metadata.fetch();
final AtomicInteger resultCount = new AtomicInteger();
final CompletableFuture<Map<TopicIdPartition, Acknowledgements>>
future = new CompletableFuture<>();
final ResultHandler resultHandler = new ResultHandler(resultCount,
Optional.of(future));
Map<Integer, Map<TopicIdPartition, Acknowledgements>>
acknowledgementsMapAllNodes = new HashMap<>();
- Map<TopicIdPartition, Acknowledgements> acknowledgementsMapCannotSend
= new HashMap<>();
- Map<TopicIdPartition, Errors> acknowledgementsMapCannotSendErrors =
new HashMap<>();
acknowledgementsMap.forEach((tip, nodeAcks) -> {
- if ((cluster.nodeById(nodeAcks.nodeId()) == null) ||
isLeaderKnownToHaveChanged(nodeAcks.nodeId(), tip)) {
- Acknowledgements prevAcks =
acknowledgementsMapCannotSend.putIfAbsent(tip, nodeAcks.acknowledgements());
- if (prevAcks != null) {
- prevAcks.merge(nodeAcks.acknowledgements());
- } else {
- acknowledgementsMapCannotSendErrors.put(tip,
acknowledgementsCannotBeSentError(nodeAcks.nodeId(), tip));
- }
- } else {
- Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeAcks.nodeId(), k -> new
HashMap<>());
- Acknowledgements prevAcks = acksMap.putIfAbsent(tip,
nodeAcks.acknowledgements());
- if (prevAcks != null) {
- prevAcks.merge(nodeAcks.acknowledgements());
- }
+ Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeAcks.nodeId(), k -> new
HashMap<>());
+ Acknowledgements prevAcks = acksMap.putIfAbsent(tip,
nodeAcks.acknowledgements());
+ if (prevAcks != null) {
+ prevAcks.merge(nodeAcks.acknowledgements());
}
});
- resultCount.addAndGet(acknowledgementsMapCannotSend.size());
-
sessionHandlers.forEach((nodeId, sessionHandler) -> {
- Map<TopicIdPartition, Acknowledgements> nodeAcknowledgements =
acknowledgementsMapAllNodes.get(nodeId);
+ Map<TopicIdPartition, Acknowledgements> nodeAcknowledgements =
acknowledgementsMapAllNodes.remove(nodeId);
if (nodeAcknowledgements == null)
return;
@@ -657,21 +699,14 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
acknowledgeRequestStates.putIfAbsent(nodeId, new Tuple<>(null,
null, null));
// Add the incoming commitSync() request to the queue.
- for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
- Acknowledgements acknowledgements =
nodeAcknowledgements.remove(tip);
- if (acknowledgements != null) {
- acknowledgementsMapToSend.put(tip, acknowledgements);
+ nodeAcknowledgements.forEach((tip, acks) -> {
+ if (acks != null) {
+ acknowledgementsMapToSend.put(tip, acks);
resultCount.incrementAndGet();
-
metricsManager.recordAcknowledgementSent(acknowledgements.size());
+ metricsManager.recordAcknowledgementSent(acks.size());
log.debug("Added sync acknowledge request for partition {}
to node {}", tip.topicPartition(), nodeId);
}
- }
-
- resultCount.addAndGet(nodeAcknowledgements.size());
- nodeAcknowledgements.forEach((tip, acks) -> {
- acks.complete(Errors.NOT_LEADER_OR_FOLLOWER.exception());
- resultHandler.complete(tip, acks,
AcknowledgeRequestType.COMMIT_SYNC, true, Optional.empty());
});
if (!acknowledgementsMapToSend.isEmpty()) {
@@ -689,10 +724,17 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
}
});
- acknowledgementsMapCannotSend.forEach((tip, acks) -> {
-
acks.complete(acknowledgementsMapCannotSendErrors.get(tip).exception());
- resultHandler.complete(tip, acks,
AcknowledgeRequestType.COMMIT_SYNC, true, Optional.empty());
- });
+ if (!acknowledgementsMapAllNodes.isEmpty()) {
+ acknowledgementsMapAllNodes.forEach((nodeId, nodeAcknowledgements)
-> resultCount.addAndGet(nodeAcknowledgements.size()));
+
+ acknowledgementsMapAllNodes.forEach((nodeId, nodeAcknowledgements)
->
+ nodeAcknowledgements.forEach((tip, acks) -> {
+ log.debug("No share session handler for node {}, failing
acknowledgements for partition {}", nodeId, tip);
+ acks.complete(acknowledgementsCannotBeSentError(nodeId,
tip).exception());
+ resultHandler.complete(tip, acks,
AcknowledgeRequestType.COMMIT_SYNC, true, Optional.empty());
+ })
+ );
+ }
resultHandler.completeIfEmpty();
return future;
@@ -708,26 +750,19 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
public void commitAsync(
final Map<TopicIdPartition, NodeAcknowledgements>
acknowledgementsMap,
final long deadlineMs) {
- final Cluster cluster = metadata.fetch();
final ResultHandler resultHandler = new
ResultHandler(Optional.empty());
Map<Integer, Map<TopicIdPartition, Acknowledgements>>
acknowledgementsMapAllNodes = new HashMap<>();
acknowledgementsMap.forEach((tip, nodeAcks) -> {
- if ((cluster.nodeById(nodeAcks.nodeId()) == null) ||
isLeaderKnownToHaveChanged(nodeAcks.nodeId(), tip)) {
- log.debug("Leader for the partition is down or has changed,
failing acknowledgements for partition {}", tip);
-
nodeAcks.acknowledgements().complete(acknowledgementsCannotBeSentError(nodeAcks.nodeId(),
tip).exception());
- maybeSendShareAcknowledgementEvent(Map.of(tip,
nodeAcks.acknowledgements()), true, Optional.empty());
- } else {
- Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeAcks.nodeId(), k -> new
HashMap<>());
- Acknowledgements prevAcks = acksMap.putIfAbsent(tip,
nodeAcks.acknowledgements());
- if (prevAcks != null) {
- prevAcks.merge(nodeAcks.acknowledgements());
- }
+ Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeAcks.nodeId(), k -> new
HashMap<>());
+ Acknowledgements prevAcks = acksMap.putIfAbsent(tip,
nodeAcks.acknowledgements());
+ if (prevAcks != null) {
+ prevAcks.merge(nodeAcks.acknowledgements());
}
});
sessionHandlers.forEach((nodeId, sessionHandler) -> {
- Map<TopicIdPartition, Acknowledgements> nodeAcknowledgements =
acknowledgementsMapAllNodes.get(nodeId);
+ Map<TopicIdPartition, Acknowledgements> nodeAcknowledgements =
acknowledgementsMapAllNodes.remove(nodeId);
if (nodeAcknowledgements == null)
return;
@@ -735,13 +770,13 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
acknowledgeRequestStates.putIfAbsent(nodeId, new Tuple<>(null,
null, null));
- for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
- Acknowledgements acknowledgements =
nodeAcknowledgements.remove(tip);
- if (acknowledgements != null) {
- acknowledgementsMapForNode.put(tip, acknowledgements);
+ nodeAcknowledgements.forEach((tip, acks) -> {
+ if (acks != null) {
+ acknowledgementsMapForNode.put(tip, acks);
-
metricsManager.recordAcknowledgementSent(acknowledgements.size());
+ metricsManager.recordAcknowledgementSent(acks.size());
log.debug("Added async acknowledge request for partition
{} to node {}", tip.topicPartition(), nodeId);
+
AcknowledgeRequestState asyncRequestState =
acknowledgeRequestStates.get(nodeId).getAsyncRequest();
if (asyncRequestState == null) {
acknowledgeRequestStates.get(nodeId).setAsyncRequest(new
AcknowledgeRequestState(logContext,
@@ -756,19 +791,22 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
AcknowledgeRequestType.COMMIT_ASYNC
));
} else {
- Acknowledgements prevAcks =
asyncRequestState.acknowledgementsToSend.putIfAbsent(tip, acknowledgements);
+ Acknowledgements prevAcks =
asyncRequestState.acknowledgementsToSend.putIfAbsent(tip, acks);
if (prevAcks != null) {
-
asyncRequestState.acknowledgementsToSend.get(tip).merge(acknowledgements);
+
asyncRequestState.acknowledgementsToSend.get(tip).merge(acks);
}
}
}
- }
+ });
+ });
+ acknowledgementsMapAllNodes.forEach((nodeId, nodeAcknowledgements) ->
nodeAcknowledgements.forEach((tip, acks) -> {
- acks.complete(Errors.NOT_LEADER_OR_FOLLOWER.exception());
+ log.debug("No share session handler for node {}, failing
acknowledgements for partition {}", nodeId, tip);
+ acks.complete(acknowledgementsCannotBeSentError(nodeId,
tip).exception());
maybeSendShareAcknowledgementEvent(Map.of(tip, acks), true,
Optional.empty());
- });
- });
+ })
+ );
resultHandler.completeIfEmpty();
}
@@ -786,7 +824,6 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
public CompletableFuture<Void> acknowledgeOnClose(
final Map<TopicIdPartition, NodeAcknowledgements>
acknowledgementsMap,
final long deadlineMs) {
- final Cluster cluster = metadata.fetch();
final AtomicInteger resultCount = new AtomicInteger();
final ResultHandler resultHandler = new ResultHandler(resultCount,
Optional.empty());
@@ -794,36 +831,26 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
Map<Integer, Map<TopicIdPartition, Acknowledgements>>
acknowledgementsMapAllNodes = new HashMap<>();
acknowledgementsMap.forEach((tip, nodeAcks) -> {
- if ((cluster.nodeById(nodeAcks.nodeId()) == null) ||
isLeaderKnownToHaveChanged(nodeAcks.nodeId(), tip)) {
-
nodeAcks.acknowledgements().complete(acknowledgementsCannotBeSentError(nodeAcks.nodeId(),
tip).exception());
- maybeSendShareAcknowledgementEvent(Map.of(tip,
nodeAcks.acknowledgements()), true, Optional.empty());
- } else {
- Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeAcks.nodeId(), k -> new
HashMap<>());
- Acknowledgements prevAcks = acksMap.putIfAbsent(tip,
nodeAcks.acknowledgements());
- if (prevAcks != null) {
- prevAcks.merge(nodeAcks.acknowledgements());
- }
+ Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeAcks.nodeId(), k -> new
HashMap<>());
+ Acknowledgements prevAcks = acksMap.putIfAbsent(tip,
nodeAcks.acknowledgements());
+ if (prevAcks != null) {
+ prevAcks.merge(nodeAcks.acknowledgements());
}
});
// Add any waiting piggyback acknowledgements.
fetchAcknowledgementsToSend.forEach((nodeId, nodeAcks) ->
nodeAcks.forEach((tip, acks) -> {
- if ((cluster.nodeById(nodeId) == null) ||
isLeaderKnownToHaveChanged(nodeId, tip)) {
- acks.complete(acknowledgementsCannotBeSentError(nodeId,
tip).exception());
- maybeSendShareAcknowledgementEvent(Map.of(tip, acks),
true, Optional.empty());
- } else {
- Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeId, k -> new HashMap<>());
- Acknowledgements prevAcks = acksMap.putIfAbsent(tip, acks);
- if (prevAcks != null) {
- prevAcks.merge(acks);
- }
+ Map<TopicIdPartition, Acknowledgements> acksMap =
acknowledgementsMapAllNodes.computeIfAbsent(nodeId, k -> new HashMap<>());
+ Acknowledgements prevAcks = acksMap.putIfAbsent(tip, acks);
+ if (prevAcks != null) {
+ prevAcks.merge(acks);
}
})
);
sessionHandlers.forEach((nodeId, sessionHandler) -> {
- Map<TopicIdPartition, Acknowledgements> nodeAcknowledgements =
acknowledgementsMapAllNodes.get(nodeId);
+ Map<TopicIdPartition, Acknowledgements> nodeAcknowledgements =
acknowledgementsMapAllNodes.remove(nodeId);
if (nodeAcknowledgements != null) {
nodeAcknowledgements.forEach((tip, acknowledgements) -> {
resultCount.incrementAndGet();
@@ -863,33 +890,18 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
}
});
+ acknowledgementsMapAllNodes.forEach((nodeId, nodeAcknowledgements) ->
+ nodeAcknowledgements.forEach((tip, acks) -> {
+ log.debug("No share session handler for node {}, failing
acknowledgements for partition {}", nodeId, tip);
+ acks.complete(acknowledgementsCannotBeSentError(nodeId,
tip).exception());
+ maybeSendShareAcknowledgementEvent(Map.of(tip, acks), true,
Optional.empty());
+ })
+ );
+
resultHandler.completeIfEmpty();
return closeFuture;
}
- /**
- * The method checks whether the leader for a topicIdPartition has changed.
- *
- * @param nodeId The previous leader for the partition.
- * @param topicIdPartition The TopicIdPartition to check.
- * @return Returns true if leader information is available and leader has
changed.
- * If the leader information is not available or if the leader has not
changed, it returns false.
- */
- private boolean isLeaderKnownToHaveChanged(int nodeId, TopicIdPartition
topicIdPartition) {
- Optional<Node> leaderNode =
metadata.currentLeader(topicIdPartition.topicPartition()).leader;
- if (leaderNode.isPresent()) {
- if (leaderNode.get().id() != nodeId) {
- log.debug("Node {} is no longer the leader for partition {},
failing acknowledgements", nodeId, topicIdPartition);
- return true;
- }
- } else {
- log.debug("No leader found for partition {}", topicIdPartition);
- metadata.requestUpdate(false);
- return false;
- }
- return false;
- }
-
private void handleShareFetchSuccess(Node fetchTarget,
ShareFetchRequestData requestData,
ClientResponse resp) {
@@ -1028,6 +1040,8 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
fetchRecordsNodeId.compareAndSet(fetchTarget.id(), -1);
}
nodesWithPendingRequests.remove(fetchTarget.id());
+
+ maybeRemoveSessionHandlerOnClose(fetchTarget.id(), requestData);
}
}
@@ -1068,6 +1082,19 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
fetchRecordsNodeId.compareAndSet(fetchTarget.id(), -1);
}
nodesWithPendingRequests.remove(fetchTarget.id());
+
+ maybeRemoveSessionHandlerOnClose(fetchTarget.id(), requestData);
+ }
+ }
+
+ /**
+ * If the completed ShareFetch was closing the share session with a final
epoch, the session handler
+ * is no longer quired. A subsequent poll will establish a new session if
partitions return to the node.
+ */
+ private void maybeRemoveSessionHandlerOnClose(int nodeId,
ShareFetchRequestData requestData) {
+ if (requestData.shareSessionEpoch() ==
ShareRequestMetadata.FINAL_EPOCH) {
+ log.debug("Removing session handler for node {} after closing
empty share session", nodeId);
+ sessionHandlers.remove(nodeId);
}
}
@@ -1269,7 +1296,27 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
* Otherwise, the error is {@link Errors#NETWORK_EXCEPTION}.
*/
private Errors acknowledgementsCannotBeSentError(int nodeId,
TopicIdPartition tip) {
- return isLeaderKnownToHaveChanged(nodeId, tip) ?
Errors.NOT_LEADER_OR_FOLLOWER : Errors.NETWORK_EXCEPTION;
+ LeaderIdAndEpoch leaderAndEpoch = shareSessionLeaderMap.get(tip);
+ if (leaderAndEpoch != null &&
metadata.fetch().nodeById(leaderAndEpoch.leaderId) != null) {
+ if (leaderAndEpoch.leaderId != nodeId) {
+ log.debug("Node {} is no longer the leader for partition {},
failing acknowledgements", nodeId, tip);
+ return Errors.NOT_LEADER_OR_FOLLOWER;
+ }
+ return Errors.NETWORK_EXCEPTION;
+ }
+
+ Optional<Node> leaderNode =
metadata.currentLeader(tip.topicPartition()).leader;
+ if (leaderNode.isPresent()) {
+ if (leaderNode.get().id() != nodeId) {
+ log.debug("Node {} is no longer the leader for partition {},
failing acknowledgements", nodeId, tip);
+ return Errors.NOT_LEADER_OR_FOLLOWER;
+ }
+ } else {
+ log.debug("No leader found for partition {}", tip);
+ metadata.requestUpdate(false);
+ }
+
+ return Errors.NETWORK_EXCEPTION;
}
private TopicIdPartition lookupTopicId(Uuid topicId, int partitionIndex) {
@@ -1403,7 +1450,7 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
incompleteAcknowledgements.isEmpty() ? acknowledgementsToSend
: incompleteAcknowledgements);
for (Map.Entry<TopicIdPartition, Acknowledgements> entry :
finalAcknowledgementsToSend.entrySet()) {
- sessionHandler.addPartitionToFetch(entry.getKey(),
entry.getValue());
+ sessionHandler.addPartitionToAcknowledgeOnly(entry.getKey(),
entry.getValue());
}
ShareAcknowledgeRequest.Builder requestBuilder =
sessionHandler.newShareAcknowledgeBuilder(groupId);
@@ -1694,6 +1741,11 @@ public class ShareConsumeRequestManager implements
RequestManager, MemberStateLi
return acknowledgeRequestStates.get(nodeId);
}
+ int shareSessionNodeId(TopicIdPartition tip) {
+ LeaderIdAndEpoch leader = shareSessionLeaderMap.get(tip);
+ return leader == null ? -1 : leader.leaderId;
+ }
+
static class IdAndPartition {
private final Uuid topicId;
private final int partitionIndex;
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchBuffer.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchBuffer.java
index a7fbaddd781..c0ed8275dec 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchBuffer.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchBuffer.java
@@ -52,6 +52,10 @@ public class ShareFetchBuffer implements AutoCloseable {
private final AtomicBoolean wokenUp = new AtomicBoolean(false);
private ShareCompletedFetch nextInLineFetch;
+ // Fetches which have been consumed but still have acknowledgements
outstanding, retained here after they leave
+ // the next-in-line slot so that the share session for their node is not
closed prematurely.
+ private final Set<ShareCompletedFetch> pendingAcknowledgementFetches = new
HashSet<>();
+
public ShareFetchBuffer(final LogContext logContext) {
this.log = logContext.logger(ShareFetchBuffer.class);
this.completedFetches = new ConcurrentLinkedQueue<>();
@@ -95,6 +99,11 @@ public class ShareFetchBuffer implements AutoCloseable {
void setNextInLineFetch(ShareCompletedFetch nextInLineFetch) {
lock.lock();
try {
+ // If the outgoing next-in-line fetch still has outstanding
acknowledgements,
+ // retain it until the acknowledgements are sent.
+ if (this.nextInLineFetch != null &&
this.nextInLineFetch.hasPendingAcknowledgements()) {
+ pendingAcknowledgementFetches.add(this.nextInLineFetch);
+ }
this.nextInLineFetch = nextInLineFetch;
} finally {
lock.unlock();
@@ -169,7 +178,7 @@ public class ShareFetchBuffer implements AutoCloseable {
}
/**
- * Return the set of {@link TopicIdPartition partitions} for which we have
data in the buffer.
+ * Return the set of {@link TopicIdPartition partitions} for which we have
data in the buffer or pending acknowledgements.
*
* @return {@link TopicIdPartition Partition} set
*/
@@ -178,17 +187,50 @@ public class ShareFetchBuffer implements AutoCloseable {
try {
final Set<TopicIdPartition> partitions = new HashSet<>();
- if (nextInLineFetch != null && !nextInLineFetch.isConsumed()) {
+ if (nextInLineFetch != null && (!nextInLineFetch.isConsumed() ||
nextInLineFetch.hasPendingAcknowledgements())) {
partitions.add(nextInLineFetch.partition);
}
completedFetches.forEach(cf -> partitions.add(cf.partition));
+
+ prunePendingAcknowledgementFetches();
+ pendingAcknowledgementFetches.forEach(cf ->
partitions.add(cf.partition));
+
return partitions;
} finally {
lock.unlock();
}
}
+ /**
+ * Return the set of node IDs for which we have data in the buffer.
+ *
+ * @return Node ID set
+ */
+ Set<Integer> bufferedNodes() {
+ lock.lock();
+ try {
+ final Set<Integer> nodes = new HashSet<>();
+
+ if (nextInLineFetch != null && (!nextInLineFetch.isConsumed() ||
nextInLineFetch.hasPendingAcknowledgements())) {
+ nodes.add(nextInLineFetch.nodeId);
+ }
+
+ completedFetches.forEach(cf -> nodes.add(cf.nodeId));
+
+ prunePendingAcknowledgementFetches();
+ pendingAcknowledgementFetches.forEach(cf -> nodes.add(cf.nodeId));
+
+ return nodes;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void prunePendingAcknowledgementFetches() {
+ pendingAcknowledgementFetches.removeIf(cf ->
!cf.hasPendingAcknowledgements());
+ }
+
private void drainAll() {
lock.lock();
try {
@@ -198,6 +240,8 @@ public class ShareFetchBuffer implements AutoCloseable {
nextInLineFetch.drain();
nextInLineFetch = null;
}
+ pendingAcknowledgementFetches.forEach(ShareCompletedFetch::drain);
+ pendingAcknowledgementFetches.clear();
} finally {
lock.unlock();
}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
index c89716ad622..59caf25c083 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java
@@ -41,7 +41,6 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -94,10 +93,6 @@ public class ShareSessionHandler {
return sessionPartitions;
}
- public Collection<TopicIdPartition> sessionPartitions() {
- return Set.copyOf(sessionPartitions.values());
- }
-
public void addPartitionToFetch(TopicIdPartition topicIdPartition,
Acknowledgements partitionAcknowledgements) {
nextPartitions.put(topicIdPartition.topicPartition(),
topicIdPartition);
if (partitionAcknowledgements != null) {
@@ -113,6 +108,10 @@ public class ShareSessionHandler {
return nextMetadata.isNewSession();
}
+ public boolean isSessionEmpty() {
+ return sessionPartitions.isEmpty();
+ }
+
public ShareFetchRequest.Builder newShareFetchBuilder(String groupId,
ShareFetchConfig shareFetchConfig, boolean canSkipIfRequestEmpty) {
List<TopicIdPartition> added = new ArrayList<>();
List<TopicIdPartition> removed = new ArrayList<>();
@@ -190,7 +189,7 @@ public class ShareSessionHandler {
nextAcknowledgements = new LinkedHashMap<>();
// If there are no changes to the share session and no
acknowledgements, we can sometimes skip sending an empty request
- if (added.isEmpty() && removed.isEmpty() &&
acknowledgementBatches.isEmpty()) {
+ if (added.isEmpty() && removed.isEmpty() &&
acknowledgementBatches.isEmpty() && !nextMetadata.isFinalEpoch()) {
// If the share session is empty, there are no partitions to fetch
from and we can always skip
if (sessionPartitions.isEmpty()) {
log.debug("Skipping sending empty ShareFetch because share
partitions empty");
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetchTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetchTest.java
index a6bc4943821..4bbf8ea08cf 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetchTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareCompletedFetchTest.java
@@ -23,6 +23,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.errors.CorruptRecordException;
+import org.apache.kafka.common.errors.NetworkException;
import org.apache.kafka.common.errors.RecordDeserializationException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
@@ -58,6 +59,7 @@ import java.util.UUID;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -110,6 +112,76 @@ public class ShareCompletedFetchTest {
assertEquals(DEFAULT_ACQUISITION_LOCK_TIMEOUT_MS,
batch.getAcquisitionLockTimeoutMs());
}
+ @Test
+ public void testHasPendingAcknowledgements() {
+ long startingOffset = 10L;
+ int numRecords = 5;
+ ShareFetchResponseData.PartitionData partitionData = new
ShareFetchResponseData.PartitionData()
+ .setRecords(newRecords(startingOffset, numRecords, 1))
+ .setAcquiredRecords(acquiredRecords(startingOffset, numRecords));
+
+ Deserializers<String, String> deserializers = newStringDeserializers();
+ ShareCompletedFetch completedFetch =
newShareCompletedFetch(partitionData);
+
+ // No records have been delivered to the application yet.
+ assertFalse(completedFetch.hasPendingAcknowledgements());
+
+ // Once the records have been delivered, their acknowledgements are
outstanding even though the fetch is consumed.
+ ShareInFlightBatch<String, String> batch =
completedFetch.fetchRecords(deserializers, 10, true);
+ assertEquals(numRecords, batch.getInFlightRecords().size());
+ assertTrue(completedFetch.isConsumed());
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+
+ // Acknowledging the records and taking them to be sent by the
background thread, clears the outstanding acknowledgements.
+ batch.acknowledgeAll(AcknowledgeType.ACCEPT);
+ batch.takeAcknowledgedRecords();
+ assertFalse(completedFetch.hasPendingAcknowledgements());
+ }
+
+ @Test
+ public void testPendingRenewAcknowledgementsWithException() {
+ long startingOffset = 10L;
+ int numRecords = 5;
+ ShareFetchResponseData.PartitionData partitionData = new
ShareFetchResponseData.PartitionData()
+ .setRecords(newRecords(startingOffset, numRecords, 1))
+ .setAcquiredRecords(acquiredRecords(startingOffset, numRecords));
+
+ Deserializers<String, String> deserializers = newStringDeserializers();
+ ShareCompletedFetch completedFetch =
newShareCompletedFetch(partitionData);
+
+ // No records have been delivered to the application yet.
+ assertFalse(completedFetch.hasPendingAcknowledgements());
+
+ // Once the record have been delivered, their acknowledgements are
outstanding even though the fetch is consumed.
+ ShareInFlightBatch<String, String> batch =
completedFetch.fetchRecords(deserializers, 10, true);
+ assertEquals(numRecords, batch.getInFlightRecords().size());
+ assertTrue(completedFetch.isConsumed());
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+
+ // Acknowledging the records and taking them to be sent by the
background thread, leaves the acknowledgements outstanding.
+ batch.acknowledgeAll(AcknowledgeType.RENEW);
+ Acknowledgements renewAcknowledgements =
batch.takeAcknowledgedRecords();
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+
+ renewAcknowledgements.complete(null);
+ batch.renew(renewAcknowledgements);
+ assertTrue(completedFetch.isConsumed());
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+
+ batch.takeRenewals();
+ assertTrue(completedFetch.isConsumed());
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+
+ batch.acknowledgeAll(AcknowledgeType.RENEW);
+ renewAcknowledgements = batch.takeAcknowledgedRecords();
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+
+ renewAcknowledgements.complete(new NetworkException());
+ batch.renew(renewAcknowledgements);
+ assertTrue(completedFetch.isConsumed());
+ assertFalse(completedFetch.hasPendingAcknowledgements());
+ }
+
@Test
public void testSoftMaxPollRecordLimit() {
long startingOffset = 10L;
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
index ce756395a37..76e71f86837 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
@@ -65,9 +65,11 @@ import org.apache.kafka.common.record.internal.SimpleRecord;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.requests.RequestHeader;
import org.apache.kafka.common.requests.RequestTestUtils;
+import org.apache.kafka.common.requests.ShareAcknowledgeRequest;
import org.apache.kafka.common.requests.ShareAcknowledgeResponse;
import org.apache.kafka.common.requests.ShareFetchRequest;
import org.apache.kafka.common.requests.ShareFetchResponse;
+import org.apache.kafka.common.requests.ShareRequestMetadata;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
@@ -86,6 +88,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
@@ -965,6 +968,100 @@ public class ShareConsumeRequestManagerTest {
assertTrue(shareConsumeRequestManager.hasCompletedFetches());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testCommitWithRenewAcknowledgements(boolean commitSync) {
+ buildRequestManager();
+
shareConsumeRequestManager.setAcknowledgementCommitCallbackRegistered(true);
+
+ assignFromSubscribed(Set.of(tp0));
+ sendFetchAndVerifyResponse(records, acquiredRecords, Errors.NONE);
+ ShareFetch<byte[], byte[]> fetch = collectFetch();
+ assertEquals(3, fetch.numRecords());
+
+ int nodeId0 = metadata.fetch().leaderFor(tp0).id();
+
+ // The application renews the delivered records. They are still
logically held by the consumer.
+ fetch.acknowledgeAll(AcknowledgeType.RENEW);
+ Map<TopicIdPartition, NodeAcknowledgements> renewAcknowledgements =
fetch.takeAcknowledgedRecords();
+ Acknowledgements acknowledgements =
renewAcknowledgements.get(tip0).acknowledgements();
+
+ // Renew acknowledgements are committed through a ShareAcknowledge
request (not piggybacked on a ShareFetch).
+ CompletableFuture<Map<TopicIdPartition, Acknowledgements>> future =
null;
+ if (commitSync) {
+ future =
shareConsumeRequestManager.commitSync(renewAcknowledgements,
calculateDeadlineMs(time.timer(2000)));
+ } else {
+ shareConsumeRequestManager.commitAsync(renewAcknowledgements,
calculateDeadlineMs(time.timer(defaultApiTimeoutMs)));
+ }
+
+ // The partition is no longer assigned, so the share session would
normally be tidied up and eventually closed.
+ // However, the delivery of the records with renew acknowledgements
must be completed before that can happen.
+ subscriptions.assignFromSubscribed(Set.of());
+
+ NetworkClientDelegate.PollResult pollResult =
shareConsumeRequestManager.poll(time.milliseconds());
+ assertEquals(1, pollResult.unsentRequests.size());
+ ShareAcknowledgeRequest.Builder builder =
(ShareAcknowledgeRequest.Builder)
pollResult.unsentRequests.get(0).requestBuilder();
+ assertTrue(builder.data().isRenewAck());
+ networkClientDelegate.addAll(pollResult.unsentRequests);
+ assertEquals(0, renewedRecords.size());
+
+ // While the renew acknowledgements are in flight, the share session
is not closed.
+ assertEquals(0, sendFetches());
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ // The broker responds to the renew acknowledgements and the records
are reported to the application.
+ client.prepareResponse(fullAcknowledgeResponse(tip0, Errors.NONE));
+ networkClientDelegate.poll(time.timer(0));
+ assertEquals(Map.of(tip0, acknowledgements),
completedAcknowledgements.get(0));
+ assertEquals(Set.of(1L, 2L, 3L), renewedRecords);
+ if (commitSync) {
+ assertTrue(future.isDone());
+ }
+
+ // The revoked partition is removed from the share session, but the
session cannot be closed while the records are still held.
+ NetworkClientDelegate.PollResult removeResult =
shareConsumeRequestManager.sendFetchesReturnPollResult();
+ assertEquals(1, removeResult.unsentRequests.size());
+ ShareFetchRequest.Builder removeBuilder = (ShareFetchRequest.Builder)
removeResult.unsentRequests.get(0).requestBuilder();
+ assertNotEquals(ShareRequestMetadata.FINAL_EPOCH,
removeBuilder.data().shareSessionEpoch());
+ assertEquals(1, removeBuilder.data().forgottenTopicsData().size());
+ assertEquals(tip0.topicId(),
removeBuilder.data().forgottenTopicsData().get(0).topicId());
+ client.prepareResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0));
+ networkClientDelegate.poll(time.timer(0));
+
+ assertEquals(0,
shareConsumeRequestManager.sendFetchesReturnPollResult().unsentRequests.size());
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ fetch.renew(Map.of(tip0, acknowledgements), Optional.empty());
+ fetch.takeRenewedRecords();
+ assertEquals(3, fetch.numRecords());
+ assertEquals(0,
shareConsumeRequestManager.sendFetchesReturnPollResult().unsentRequests.size());
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ // The application finally accepts the records and the
acknowledgements are sent.
+ fetch.acknowledgeAll(AcknowledgeType.ACCEPT);
+ shareConsumeRequestManager.fetch(fetch.takeAcknowledgedRecords());
+ NetworkClientDelegate.PollResult ackResult =
shareConsumeRequestManager.sendFetchesReturnPollResult();
+ assertEquals(1, ackResult.unsentRequests.size());
+
+ ShareFetchRequest.Builder ackBuilder = (ShareFetchRequest.Builder)
ackResult.unsentRequests.get(0).requestBuilder();
+ assertNotEquals(ShareRequestMetadata.FINAL_EPOCH,
ackBuilder.data().shareSessionEpoch());
+ client.prepareResponse(fullFetchResponse(tip0, MemoryRecords.EMPTY,
emptyAcquiredRecords, Errors.NONE));
+ networkClientDelegate.poll(time.timer(0));
+
+ // The application consumes the fetch response corresponding to the
piggybacked acknowledgements.
+ fetchRecords();
+
+ // With the record finally acknowledged, the empty share session is
closed.
+ NetworkClientDelegate.PollResult closeResult =
shareConsumeRequestManager.sendFetchesReturnPollResult();
+ assertEquals(1, closeResult.unsentRequests.size());
+ ShareFetchRequest.Builder closeBuilder = (ShareFetchRequest.Builder)
closeResult.unsentRequests.get(0).requestBuilder();
+ assertEquals(ShareRequestMetadata.FINAL_EPOCH,
closeBuilder.data().shareSessionEpoch());
+ client.prepareResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0));
+ networkClientDelegate.poll(time.timer(0));
+
+ assertNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+ }
+
@Test
public void testCloseWithSubscriptionChange() {
buildRequestManager();
@@ -1179,6 +1276,123 @@ public class ShareConsumeRequestManagerTest {
assertEquals(0, shareConsumeRequestManager.sendFetches());
}
+ @Test
+ public void testCloseSessionWhenNoSharePartitions() {
+ buildRequestManager();
+
+ assignFromSubscribed(Set.of(tp0));
+
+ // Establish the share session by fetching from tp0 and consume the
(empty) buffered fetch.
+ sendFetchAndVerifyResponse(records, emptyAcquiredRecords, Errors.NONE);
+ fetchRecords();
+
+ Node node0 = metadata.fetch().leaderFor(tp0);
+ int nodeId0 = node0.id();
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ // The partition is no longer assigned, so a ShareFetch is built to
remove tip0 from the share session,
+ // leaving it empty.
+ subscriptions.assignFromSubscribed(Set.of());
+ assertEquals(1, sendFetches());
+ client.prepareResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0));
+ networkClientDelegate.poll(time.timer(0));
+
+ // The next poll closes the now-empty share session using a
final-epoch ShareFetch.
+ NetworkClientDelegate.PollResult pollResult =
shareConsumeRequestManager.sendFetchesReturnPollResult();
+ assertEquals(1, pollResult.unsentRequests.size());
+ assertEquals(node0, pollResult.unsentRequests.get(0).node().get());
+ ShareFetchRequest.Builder builder = (ShareFetchRequest.Builder)
pollResult.unsentRequests.get(0).requestBuilder();
+ assertEquals(ShareRequestMetadata.FINAL_EPOCH,
builder.data().shareSessionEpoch());
+ assertTrue(builder.data().topics().isEmpty());
+ assertTrue(builder.data().forgottenTopicsData().isEmpty());
+
+ // The session handle is only removed once the close response is
received.
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+ client.prepareResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0));
+ networkClientDelegate.poll(time.timer(0));
+ assertNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ // With the session closed, there is nothing left to send.
+ assertEquals(0, shareConsumeRequestManager.sendFetches());
+ }
+
+ @Test
+ public void testDoesNotCloseSessionWhileRecordsBuffered() {
+ buildRequestManager();
+
+ assignFromSubscribed(Set.of(tp0));
+
+ // Establish the share session, leaving the fetched records unconsumed
in the buffer.
+ sendFetchAndVerifyResponse(records, acquiredRecords, Errors.NONE);
+ Node node0 = metadata.fetch().leaderFor(tp0);
+ int nodeId0 = node0.id();
+
+ // Remove the partition from the session so the session becomes empty.
+ subscriptions.assignFromSubscribed(Set.of());
+ assertEquals(1, sendFetches());
+ client.prepareResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0));
+ networkClientDelegate.poll(time.timer(0));
+
+ // The session is empty, but records for the node remain in the
buffer, so the session is not closed.
+ assertEquals(0,
shareConsumeRequestManager.sendFetchesReturnPollResult().unsentRequests.size());
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ // Consuming the records hands them to the application, but leaves the
acknowledgements outstanding and the session
+ // is not closed.
+ ShareFetch<byte[], byte[]> fetch = collectFetch();
+ assertEquals(0,
shareConsumeRequestManager.sendFetchesReturnPollResult().unsentRequests.size());
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ // Acknowledge the records and send the acknowledgements. This is not
a close request.
+ fetch.acknowledgeAll(AcknowledgeType.ACCEPT);
+ shareConsumeRequestManager.fetch(fetch.takeAcknowledgedRecords());
+ NetworkClientDelegate.PollResult ackResult =
shareConsumeRequestManager.sendFetchesReturnPollResult();
+ assertEquals(1, ackResult.unsentRequests.size());
+ ShareFetchRequest.Builder ackBuilder = (ShareFetchRequest.Builder)
ackResult.unsentRequests.get(0).requestBuilder();
+ assertNotEquals(ShareRequestMetadata.FINAL_EPOCH,
ackBuilder.data().shareSessionEpoch());
+ client.prepareResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0));
+ networkClientDelegate.poll(time.timer(0));
+
+ // Once the acknowledgements have been sent, the empty session is
closed.
+ NetworkClientDelegate.PollResult closeResult =
shareConsumeRequestManager.sendFetchesReturnPollResult();
+ assertEquals(1, closeResult.unsentRequests.size());
+ assertEquals(node0, closeResult.unsentRequests.get(0).node().get());
+ ShareFetchRequest.Builder builder = (ShareFetchRequest.Builder)
closeResult.unsentRequests.get(0).requestBuilder();
+ assertEquals(ShareRequestMetadata.FINAL_EPOCH,
builder.data().shareSessionEpoch());
+ }
+
+ @Test
+ public void testDoesNotCloseSessionWhileAcknowledgementsPending() {
+ buildRequestManager();
+
+ assignFromSubscribed(Set.of(tp0));
+
+ // Establish the share session by fetching from tp0 and consume the
(empty) buffered fetch.
+ sendFetchAndVerifyResponse(records, emptyAcquiredRecords, Errors.NONE);
+ fetchRecords();
+
+ Node node0 = metadata.fetch().leaderFor(tp0);
+ int nodeId0 = node0.id();
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+
+ // Remove the partition from the session so the session becomes empty.
+ subscriptions.assignFromSubscribed(Set.of());
+ assertEquals(1, sendFetches());
+ client.prepareResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0));
+ networkClientDelegate.poll(time.timer(0));
+
+ // There are acknowledgements still to be delivered for the node, so
the session is not closed.
+ Acknowledgements acknowledgements = getAcknowledgements(0,
AcknowledgeType.ACCEPT, AcknowledgeType.ACCEPT, AcknowledgeType.ACCEPT);
+ shareConsumeRequestManager.fetch(Map.of(tip0, new
NodeAcknowledgements(nodeId0, acknowledgements)));
+
+ NetworkClientDelegate.PollResult pollResult =
shareConsumeRequestManager.poll(time.milliseconds());
+ assertEquals(1, pollResult.unsentRequests.size());
+ assertEquals(node0, pollResult.unsentRequests.get(0).node().get());
+ ShareFetchRequest.Builder builder = (ShareFetchRequest.Builder)
pollResult.unsentRequests.get(0).requestBuilder();
+ assertNotEquals(ShareRequestMetadata.FINAL_EPOCH,
builder.data().shareSessionEpoch());
+ assertNotNull(shareConsumeRequestManager.sessionHandler(nodeId0));
+ }
+
@Test
public void testShareFetchAndCloseMultipleNodes() {
buildRequestManager();
@@ -1929,7 +2143,7 @@ public class ShareConsumeRequestManagerTest {
LinkedHashMap<TopicIdPartition, ShareFetchResponseData.PartitionData>
partitionData =
buildPartitionDataMap(tip0, records,
ShareCompletedFetchTest.acquiredRecords(1L, 1), Errors.NONE, Errors.NONE);
client.prepareResponseFrom(ShareFetchResponse.of(Errors.NONE, 0,
partitionData, List.of(), 0), nodeId0);
- partitionData.clear();
+ partitionData = new LinkedHashMap<>();
partitionData.put(tip1,
new ShareFetchResponseData.PartitionData()
.setPartitionIndex(tip1.topicPartition().partition())
@@ -1996,7 +2210,7 @@ public class ShareConsumeRequestManagerTest {
*/
@ParameterizedTest
@EnumSource(value = Errors.class, names = {"FENCED_LEADER_EPOCH",
"NOT_LEADER_OR_FOLLOWER"})
- public void
testWhenFetchResponseReturnsWithALeadershipChangeErrorAndNewLeaderInformation(Errors
error) {
+ public void
testWhenShareFetchResponseReturnsWithALeadershipChangeErrorAndNewLeaderInformation(Errors
error) {
buildRequestManager();
subscriptions.subscribeToShareGroup(Set.of(topicName));
@@ -2062,12 +2276,12 @@ public class ShareConsumeRequestManagerTest {
.setAcquiredRecords(ShareCompletedFetchTest.acquiredRecords(1L, 1))
.setAcknowledgeErrorCode(Errors.NONE.code()));
client.prepareResponseFrom(ShareFetchResponse.of(Errors.NONE, 0,
partitionData, List.of(), 0), nodeId0);
- networkClientDelegate.poll(time.timer(0));
partitionData = new LinkedHashMap<>();
partitionData.put(tip1,
new ShareFetchResponseData.PartitionData()
.setPartitionIndex(tip1.topicPartition().partition()));
client.prepareResponseFrom(ShareFetchResponse.of(Errors.NONE, 0,
partitionData, List.of(), 0), nodeId1);
+ networkClientDelegate.poll(time.timer(0));
assertTrue(shareConsumeRequestManager.hasCompletedFetches());
partitionRecords = fetchRecords();
@@ -2226,26 +2440,33 @@ public class ShareConsumeRequestManagerTest {
commitAcks.put(tip0, new NodeAcknowledgements(0, acknowledgementsTp0));
commitAcks.put(tip1, new NodeAcknowledgements(1, acknowledgementsTp1));
- // Move the leadership of tp0 onto node 1
+ // Move the leadership of tp0 onto node 1.
metadata.updatePartitionLeadership(Map.of(tp0, new
Metadata.LeaderIdAndEpoch(Optional.of(nodeId1.id()),
Optional.of(validLeaderEpoch + 1))), List.of());
assertNotEquals(startingClusterMetadata, metadata.fetch());
- // We fail the acknowledgements for records which were received from
node0 with NOT_LEADER_OR_FOLLOWER exception.
+ // The acknowledgements are sent to the nodes the records were fetched
from.
shareConsumeRequestManager.commitAsync(commitAcks,
calculateDeadlineMs(time.timer(defaultApiTimeoutMs)));
- assertEquals(1, completedAcknowledgements.get(0).size());
- assertEquals(acknowledgementsTp0,
completedAcknowledgements.get(0).get(tip0));
- assertEquals(Errors.NOT_LEADER_OR_FOLLOWER.exception(),
completedAcknowledgements.get(0).get(tip0).getAcknowledgeException());
-
- // We only send acknowledgements for tip1 to node1.
- assertEquals(1, shareConsumeRequestManager.sendAcknowledgements());
+ assertTrue(completedAcknowledgements.isEmpty());
+ assertEquals(2, shareConsumeRequestManager.sendAcknowledgements());
- client.prepareResponse(fullAcknowledgeResponse(tip1, Errors.NONE));
+ // The former leader for tp0 rejects the acknowledgements with
NOT_LEADER_OR_FOLLOWER and points to the new leader.
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip0,
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ new
ShareAcknowledgeResponseData.LeaderIdAndEpoch().setLeaderId(nodeId1.id()).setLeaderEpoch(validLeaderEpoch
+ 1),
+ List.of(nodeId0, nodeId1)), nodeId0);
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip1, Errors.NONE),
nodeId1);
networkClientDelegate.poll(time.timer(0));
- assertEquals(1, completedAcknowledgements.get(1).size());
- assertEquals(acknowledgementsTp1,
completedAcknowledgements.get(1).get(tip1));
-
assertNull(completedAcknowledgements.get(1).get(tip1).getAcknowledgeException());
+ Map<TopicIdPartition, Acknowledgements> completed = new HashMap<>();
+ completedAcknowledgements.forEach(completed::putAll);
+ assertEquals(2, completed.size());
+ assertEquals(acknowledgementsTp0, completed.get(tip0));
+ assertInstanceOf(NotLeaderOrFollowerException.class,
completed.get(tip0).getAcknowledgeException());
+ assertEquals(acknowledgementsTp1, completed.get(tip1));
+ assertNull(completed.get(tip1).getAcknowledgeException());
+
+ assertEquals(nodeId1.id(),
shareConsumeRequestManager.shareSessionNodeId(tip0));
}
@Test
@@ -2296,26 +2517,32 @@ public class ShareConsumeRequestManagerTest {
commitAcks.put(tip0, new NodeAcknowledgements(0, acknowledgementsTp0));
commitAcks.put(tip1, new NodeAcknowledgements(1, acknowledgementsTp1));
- // Move the leadership of tp0 onto node 1
+ // Move the leadership of tp0 onto node 1.
metadata.updatePartitionLeadership(Map.of(tp0, new
Metadata.LeaderIdAndEpoch(Optional.of(nodeId1.id()),
Optional.of(validLeaderEpoch + 1))), List.of());
assertNotEquals(startingClusterMetadata, metadata.fetch());
- // We fail the acknowledgements for records which were received from
node0 with NOT_LEADER_OR_FOLLOWER exception.
+ // The acknowledgements are sent to the nodes the records were fetched
from.
shareConsumeRequestManager.commitSync(commitAcks,
calculateDeadlineMs(time.timer(100)));
+ assertEquals(2, shareConsumeRequestManager.sendAcknowledgements());
- // We only send acknowledgements for tip1 to node1.
- assertEquals(1, shareConsumeRequestManager.sendAcknowledgements());
-
- client.prepareResponse(fullAcknowledgeResponse(tip1, Errors.NONE));
+ // The former leader for tp0 rejects the acknowledgements with
NOT_LEADER_OR_FOLLOWER and points to the new leader.
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip0,
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ new
ShareAcknowledgeResponseData.LeaderIdAndEpoch().setLeaderId(nodeId1.id()).setLeaderEpoch(validLeaderEpoch
+ 1),
+ List.of(nodeId0, nodeId1)), nodeId0);
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip1, Errors.NONE),
nodeId1);
networkClientDelegate.poll(time.timer(0));
- // Verify if the callback was invoked with the failed
acknowledgements. The callback is called with the commitSync processing.
- assertEquals(2, completedAcknowledgements.get(0).size());
- assertEquals(acknowledgementsTp0,
completedAcknowledgements.get(0).get(tip0));
- assertEquals(Errors.NOT_LEADER_OR_FOLLOWER.exception(),
completedAcknowledgements.get(0).get(tip0).getAcknowledgeException());
- assertEquals(acknowledgementsTp1,
completedAcknowledgements.get(0).get(tip1));
-
assertNull(completedAcknowledgements.get(0).get(tip1).getAcknowledgeException());
+ Map<TopicIdPartition, Acknowledgements> completed = new HashMap<>();
+ completedAcknowledgements.forEach(completed::putAll);
+ assertEquals(2, completed.size());
+ assertEquals(acknowledgementsTp0, completed.get(tip0));
+ assertInstanceOf(NotLeaderOrFollowerException.class,
completed.get(tip0).getAcknowledgeException());
+ assertEquals(acknowledgementsTp1, completed.get(tip1));
+ assertNull(completed.get(tip1).getAcknowledgeException());
+
+ assertEquals(nodeId1.id(),
shareConsumeRequestManager.shareSessionNodeId(tip0));
}
@Test
@@ -2367,33 +2594,32 @@ public class ShareConsumeRequestManagerTest {
shareConsumeRequestManager.fetch(Map.of(tip1, new
NodeAcknowledgements(1, acknowledgementsTp1)));
- // Move the leadership of tp0 onto node 1
+ // Move the leadership of tp0 onto node 1.
metadata.updatePartitionLeadership(Map.of(tp0, new
Metadata.LeaderIdAndEpoch(Optional.of(nodeId1.id()),
Optional.of(validLeaderEpoch + 1))), List.of());
assertNotEquals(startingClusterMetadata, metadata.fetch());
- // We fail the acknowledgements for records which were received from
node0 with NOT_LEADER_OR_FOLLOWER exception.
+ // The acknowledgements are sent to the nodes the records were fetched
from.
shareConsumeRequestManager.acknowledgeOnClose(Map.of(tip0, new
NodeAcknowledgements(0, acknowledgementsTp0)),
calculateDeadlineMs(time.timer(100)));
-
- // Verify if the callback was invoked with the failed acknowledgements.
- assertEquals(1, completedAcknowledgements.get(0).size());
- assertEquals(acknowledgementsTp0.getAcknowledgementsTypeMap(),
completedAcknowledgements.get(0).get(tip0).getAcknowledgementsTypeMap());
- assertEquals(Errors.NOT_LEADER_OR_FOLLOWER.exception(),
completedAcknowledgements.get(0).get(tip0).getAcknowledgeException());
- completedAcknowledgements.clear();
-
- // As we are closing, we still send the request to both the nodes, but
with empty acknowledgements to node0, as it is no longer the leader.
+ assertTrue(completedAcknowledgements.isEmpty());
assertEquals(2, shareConsumeRequestManager.sendAcknowledgements());
+ // The former leader for tp0 rejects the acknowledgements with
NOT_LEADER_OR_FOLLOWER and points to the new leader.
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip0,
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ new
ShareAcknowledgeResponseData.LeaderIdAndEpoch().setLeaderId(nodeId1.id()).setLeaderEpoch(validLeaderEpoch
+ 1),
+ List.of(nodeId0, nodeId1)), nodeId0);
client.prepareResponseFrom(fullAcknowledgeResponse(tip1, Errors.NONE),
nodeId1);
networkClientDelegate.poll(time.timer(0));
- client.prepareResponseFrom(emptyAcknowledgeResponse(), nodeId0);
- networkClientDelegate.poll(time.timer(0));
-
- assertEquals(1, completedAcknowledgements.get(0).size());
- assertEquals(acknowledgementsTp1,
completedAcknowledgements.get(0).get(tip1));
-
assertNull(completedAcknowledgements.get(0).get(tip1).getAcknowledgeException());
+ Map<TopicIdPartition, Acknowledgements> completed = new HashMap<>();
+ completedAcknowledgements.forEach(completed::putAll);
+ assertEquals(2, completed.size());
+ assertEquals(acknowledgementsTp0, completed.get(tip0));
+ assertInstanceOf(NotLeaderOrFollowerException.class,
completed.get(tip0).getAcknowledgeException());
+ assertEquals(acknowledgementsTp1, completed.get(tip1));
+ assertNull(completed.get(tip1).getAcknowledgeException());
}
@Test
@@ -2445,33 +2671,32 @@ public class ShareConsumeRequestManagerTest {
shareConsumeRequestManager.fetch(Map.of(tip1, new
NodeAcknowledgements(1, acknowledgementsTp1)));
- // Move the leadership of tp1 onto node 0
+ // Move the leadership of tp1 onto node 0.
metadata.updatePartitionLeadership(Map.of(tp1, new
Metadata.LeaderIdAndEpoch(Optional.of(nodeId0.id()),
Optional.of(validLeaderEpoch + 1))), List.of());
assertNotEquals(startingClusterMetadata, metadata.fetch());
- // We fail the acknowledgements for records which were received from
node0 with NOT_LEADER_OR_FOLLOWER exception.
+ // The acknowledgements are sent to the nodes the records were fetched
from.
shareConsumeRequestManager.acknowledgeOnClose(Map.of(tip0, new
NodeAcknowledgements(0, acknowledgementsTp0)),
calculateDeadlineMs(time.timer(100)));
-
- // Verify if the callback was invoked with the failed acknowledgements.
- assertEquals(1, completedAcknowledgements.get(0).size());
- assertEquals(acknowledgementsTp1.getAcknowledgementsTypeMap(),
completedAcknowledgements.get(0).get(tip1).getAcknowledgementsTypeMap());
- assertEquals(Errors.NOT_LEADER_OR_FOLLOWER.exception(),
completedAcknowledgements.get(0).get(tip1).getAcknowledgeException());
- completedAcknowledgements.clear();
-
- // As we are closing, we still send the request to both the nodes, but
with empty acknowledgements to node1, as it is no longer the leader.
+ assertTrue(completedAcknowledgements.isEmpty());
assertEquals(2, shareConsumeRequestManager.sendAcknowledgements());
+ // The former leader for tp0 rejects the acknowledgements with
NOT_LEADER_OR_FOLLOWER and points to the new leader.
client.prepareResponseFrom(fullAcknowledgeResponse(tip0, Errors.NONE),
nodeId0);
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip1,
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ new
ShareAcknowledgeResponseData.LeaderIdAndEpoch().setLeaderId(nodeId0.id()).setLeaderEpoch(validLeaderEpoch
+ 1),
+ List.of(nodeId0, nodeId1)), nodeId1);
networkClientDelegate.poll(time.timer(0));
- client.prepareResponseFrom(emptyAcknowledgeResponse(), nodeId1);
- networkClientDelegate.poll(time.timer(0));
-
- assertEquals(1, completedAcknowledgements.get(0).size());
- assertEquals(acknowledgementsTp0,
completedAcknowledgements.get(0).get(tip0));
-
assertNull(completedAcknowledgements.get(0).get(tip0).getAcknowledgeException());
+ Map<TopicIdPartition, Acknowledgements> completed = new HashMap<>();
+ completedAcknowledgements.forEach(completed::putAll);
+ assertEquals(2, completed.size());
+ assertEquals(acknowledgementsTp0, completed.get(tip0));
+ assertNull(completed.get(tip0).getAcknowledgeException());
+ assertEquals(acknowledgementsTp1.getAcknowledgementsTypeMap(),
completed.get(tip1).getAcknowledgementsTypeMap());
+ assertInstanceOf(NotLeaderOrFollowerException.class,
completed.get(tip1).getAcknowledgeException());
}
@Test
@@ -2523,35 +2748,109 @@ public class ShareConsumeRequestManagerTest {
shareConsumeRequestManager.fetch(Map.of(tip1, new
NodeAcknowledgements(1, acknowledgementsTp1)));
- // Move the leadership of tp1 onto node 0, and tp0 onto node 1
+ // Move the leadership of tp1 onto node 0, and tp0 onto node 1.
metadata.updatePartitionLeadership(Map.of(tp1, new
Metadata.LeaderIdAndEpoch(Optional.of(nodeId0.id()),
Optional.of(validLeaderEpoch + 1))), List.of());
metadata.updatePartitionLeadership(Map.of(tp0, new
Metadata.LeaderIdAndEpoch(Optional.of(nodeId1.id()),
Optional.of(validLeaderEpoch + 1))), List.of());
assertNotEquals(startingClusterMetadata, metadata.fetch());
- // We fail the acknowledgements for records which were received from
node0 and node 1 with NOT_LEADER_OR_FOLLOWER exception.
+ // The acknowledgements are sent to the nodes the records were fetched
from.
shareConsumeRequestManager.acknowledgeOnClose(Map.of(tip0, new
NodeAcknowledgements(0, acknowledgementsTp0)),
calculateDeadlineMs(time.timer(100)));
-
- // Verify if the callback was invoked with the failed acknowledgements.
- assertEquals(1, completedAcknowledgements.get(0).size());
- assertEquals(acknowledgementsTp0.getAcknowledgementsTypeMap(),
completedAcknowledgements.get(0).get(tip0).getAcknowledgementsTypeMap());
- assertEquals(Errors.NOT_LEADER_OR_FOLLOWER.exception(),
completedAcknowledgements.get(0).get(tip0).getAcknowledgeException());
- assertEquals(1, completedAcknowledgements.get(1).size());
- assertEquals(acknowledgementsTp1.getAcknowledgementsTypeMap(),
completedAcknowledgements.get(1).get(tip1).getAcknowledgementsTypeMap());
- assertEquals(Errors.NOT_LEADER_OR_FOLLOWER.exception(),
completedAcknowledgements.get(1).get(tip1).getAcknowledgeException());
- completedAcknowledgements.clear();
-
- // As we are closing, we still send the request to both the nodes, but
with empty acknowledgements to node 0 and node1, as they are no longer the
leader.
assertEquals(2, shareConsumeRequestManager.sendAcknowledgements());
- client.prepareResponseFrom(emptyAcknowledgeResponse(), nodeId0);
+ // The former leaders reject the acknowledgements with
NOT_LEADER_OR_FOLLOWER and point to the new leaders.
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip0,
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ new
ShareAcknowledgeResponseData.LeaderIdAndEpoch().setLeaderId(nodeId1.id()).setLeaderEpoch(validLeaderEpoch
+ 1),
+ List.of(nodeId0, nodeId1)), nodeId0);
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip1,
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ new
ShareAcknowledgeResponseData.LeaderIdAndEpoch().setLeaderId(nodeId0.id()).setLeaderEpoch(validLeaderEpoch
+ 1),
+ List.of(nodeId0, nodeId1)), nodeId1);
networkClientDelegate.poll(time.timer(0));
- client.prepareResponseFrom(emptyAcknowledgeResponse(), nodeId1);
+ Map<TopicIdPartition, Acknowledgements> completed = new HashMap<>();
+ completedAcknowledgements.forEach(completed::putAll);
+ assertEquals(2, completed.size());
+ assertEquals(acknowledgementsTp0.getAcknowledgementsTypeMap(),
completed.get(tip0).getAcknowledgementsTypeMap());
+ assertInstanceOf(NotLeaderOrFollowerException.class,
completed.get(tip0).getAcknowledgeException());
+ assertEquals(acknowledgementsTp1.getAcknowledgementsTypeMap(),
completed.get(tip1).getAcknowledgementsTypeMap());
+ assertInstanceOf(NotLeaderOrFollowerException.class,
completed.get(tip1).getAcknowledgeException());
+ }
+
+ @Test
+ void testStaleMetadataLeadershipChangeDoesNotFailAcknowledgements() {
+ buildRequestManager();
+
shareConsumeRequestManager.setAcknowledgementCommitCallbackRegistered(true);
+
+ subscriptions.subscribeToShareGroup(Set.of(topicName));
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(tp0);
+ partitions.add(tp1);
+ subscriptions.assignFromSubscribed(partitions);
+
+ client.updateMetadata(
+ RequestTestUtils.metadataUpdateWithIds(2, Map.of(topicName, 2),
+ tp -> validLeaderEpoch, topicIds, false));
+ Node nodeId0 = metadata.fetch().nodeById(0);
+ Node nodeId1 = metadata.fetch().nodeById(1);
+
+ Cluster startingClusterMetadata = metadata.fetch();
+ assertFalse(metadata.updateRequested());
+
+ assertEquals(2, sendFetches());
+ assertFalse(shareConsumeRequestManager.hasCompletedFetches());
+
+ LinkedHashMap<TopicIdPartition, ShareFetchResponseData.PartitionData>
partitionData =
+ buildPartitionDataMap(tip0, records,
ShareCompletedFetchTest.acquiredRecords(1L, 1), Errors.NONE, Errors.NONE);
+ client.prepareResponseFrom(ShareFetchResponse.of(Errors.NONE, 0,
partitionData, List.of(), 0), nodeId0);
+ partitionData = buildPartitionDataMap(tip1, records,
ShareCompletedFetchTest.acquiredRecords(1L, 2), Errors.NONE, Errors.NONE);
+ client.prepareResponseFrom(ShareFetchResponse.of(Errors.NONE, 0,
partitionData, List.of(), 0), nodeId1);
networkClientDelegate.poll(time.timer(0));
+ assertTrue(shareConsumeRequestManager.hasCompletedFetches());
+ Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>
partitionRecords = fetchRecords();
+ assertTrue(partitionRecords.containsKey(tp0));
+ assertTrue(partitionRecords.containsKey(tp1));
+
+ Acknowledgements acknowledgementsTp0 = Acknowledgements.empty();
+ acknowledgementsTp0.add(1L, AcknowledgeType.ACCEPT);
+
+ Acknowledgements acknowledgementsTp1 = getAcknowledgements(1,
+ AcknowledgeType.ACCEPT, AcknowledgeType.ACCEPT);
+
+ Map<TopicIdPartition, NodeAcknowledgements> commitAcks = new
HashMap<>();
+ commitAcks.put(tip0, new NodeAcknowledgements(0, acknowledgementsTp0));
+ commitAcks.put(tip1, new NodeAcknowledgements(1, acknowledgementsTp1));
+
+ // The metadata reports tp0 has moved to node 1, but the cached leader
used for fetching (node 0) is unchanged.
+ metadata.updatePartitionLeadership(Map.of(tp0, new
Metadata.LeaderIdAndEpoch(Optional.of(nodeId1.id()),
Optional.of(validLeaderEpoch + 1))), List.of());
+
+ assertNotEquals(startingClusterMetadata, metadata.fetch());
+
+ // The acknowledgements are sent to the nodes the records were fetched
from, so nothing fails synchronously.
+ shareConsumeRequestManager.commitAsync(commitAcks,
calculateDeadlineMs(time.timer(defaultApiTimeoutMs)));
assertTrue(completedAcknowledgements.isEmpty());
+
+ // We send acknowledgements for tip0 to node0 and tip1 to node1.
+ assertEquals(2, shareConsumeRequestManager.sendAcknowledgements());
+
+ // Both cached leaders accept the acknowledgements despite the stale
metadata for tp0.
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip0, Errors.NONE),
nodeId0);
+ client.prepareResponseFrom(fullAcknowledgeResponse(tip1, Errors.NONE),
nodeId1);
+ networkClientDelegate.poll(time.timer(0));
+
+ Map<TopicIdPartition, Acknowledgements> completed = new HashMap<>();
+ completedAcknowledgements.forEach(completed::putAll);
+ assertEquals(2, completed.size());
+ assertEquals(acknowledgementsTp0, completed.get(tip0));
+ assertNull(completed.get(tip0).getAcknowledgeException());
+ assertEquals(acknowledgementsTp1, completed.get(tip1));
+ assertNull(completed.get(tip1).getAcknowledgeException());
+
+ // The cached leader for tp0 is unchanged - the broker success did not
repoint it to the stale metadata leader.
+ assertEquals(nodeId0.id(),
shareConsumeRequestManager.shareSessionNodeId(tip0));
}
@Test
@@ -2604,9 +2903,9 @@ public class ShareConsumeRequestManagerTest {
assertEquals(startingClusterMetadata, metadata.fetch());
- Acknowledgements acknowledgements = Acknowledgements.empty();
- acknowledgements.add(1, AcknowledgeType.ACCEPT);
- shareConsumeRequestManager.fetch(Map.of(tip0, new
NodeAcknowledgements(0, acknowledgements)));
+ Acknowledgements acknowledgements1 = Acknowledgements.empty();
+ acknowledgements1.add(1, AcknowledgeType.ACCEPT);
+ shareConsumeRequestManager.fetch(Map.of(tip0, new
NodeAcknowledgements(0, acknowledgements1)));
assertEquals(2, sendFetches());
assertFalse(shareConsumeRequestManager.hasCompletedFetches());
@@ -2645,7 +2944,9 @@ public class ShareConsumeRequestManagerTest {
assertNotEquals(startingClusterMetadata, metadata.fetch());
- shareConsumeRequestManager.fetch(Map.of(tip1, new
NodeAcknowledgements(1, acknowledgements)));
+ Acknowledgements acknowledgements2 = Acknowledgements.empty();
+ acknowledgements2.add(1, AcknowledgeType.ACCEPT);
+ shareConsumeRequestManager.fetch(Map.of(tip1, new
NodeAcknowledgements(1, acknowledgements2)));
assertEquals(2, sendFetches());
assertFalse(shareConsumeRequestManager.hasCompletedFetches());
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareFetchBufferTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareFetchBufferTest.java
index 0e76c37e11a..de757599cb0 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareFetchBufferTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareFetchBufferTest.java
@@ -16,12 +16,19 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.consumer.AcknowledgeType;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.message.ShareFetchResponseData;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.record.internal.MemoryRecords;
+import org.apache.kafka.common.record.internal.MemoryRecordsBuilder;
+import org.apache.kafka.common.record.internal.Records;
+import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
@@ -32,6 +39,7 @@ import org.apache.kafka.common.utils.internals.LogContext;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
@@ -65,6 +73,7 @@ public class ShareFetchBufferTest {
private final TopicIdPartition topicAPartition1 = new
TopicIdPartition(Uuid.randomUuid(), 1, "topic-a");
private final TopicIdPartition topicAPartition2 = new
TopicIdPartition(Uuid.randomUuid(), 2, "topic-a");
private final Set<TopicIdPartition> allPartitions =
partitions(topicAPartition0, topicAPartition1, topicAPartition2);
+ private final Deserializers<String, String> deserializers = new
Deserializers<>(new StringDeserializer(), new StringDeserializer(), null);
private LogContext logContext;
private ShareFetchMetricsManager shareFetchMetricsManager;
@@ -136,18 +145,121 @@ public class ShareFetchBufferTest {
fetchBuffer.setNextInLineFetch(completedFetch(topicAPartition0));
fetchBuffer.add(List.of(completedFetch(topicAPartition1),
completedFetch(topicAPartition2)));
assertEquals(allPartitions, fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(0), fetchBuffer.bufferedNodes());
fetchBuffer.setNextInLineFetch(null);
assertEquals(partitions(topicAPartition1, topicAPartition2),
fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(0), fetchBuffer.bufferedNodes());
fetchBuffer.poll();
assertEquals(partitions(topicAPartition2),
fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(0), fetchBuffer.bufferedNodes());
fetchBuffer.poll();
assertEquals(partitions(), fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(), fetchBuffer.bufferedNodes());
}
}
+ /**
+ * Tests that a fetch which has been consumed but whose acknowledgements
are still outstanding continues to be
+ * reported as buffered, thus preventing the share session for its node
being closed prematurely.
+ */
+ @Test
+ public void testBufferedIncludesConsumedFetchWithPendingAcknowledgements()
{
+ try (ShareFetchBuffer fetchBuffer = new ShareFetchBuffer(logContext)) {
+ ShareCompletedFetch completedFetch =
completedFetchWithAcquiredRecords(topicAPartition0, 0, 5);
+ ShareInFlightBatch<String, String> batch =
consumeFetchWithAcquiredRecords(completedFetch);
+ fetchBuffer.setNextInLineFetch(completedFetch);
+
+ // Even though the fetch has been consumed, its acknowledgements
are still outstanding, so it is still buffered.
+ assertEquals(partitions(topicAPartition0),
fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(0), fetchBuffer.bufferedNodes());
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+
+ // Acknowledging and taking the records clears the outstanding
acknowledgements, so it is no longer buffered.
+ batch.acknowledgeAll(AcknowledgeType.ACCEPT);
+ batch.takeAcknowledgedRecords();
+ assertFalse(completedFetch.hasPendingAcknowledgements());
+ assertEquals(partitions(), fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(), fetchBuffer.bufferedNodes());
+ }
+ }
+
+ @Test
+ public void testRetainsEvictedFetchWithPendingAcknowledgements() {
+ try (ShareFetchBuffer fetchBuffer = new ShareFetchBuffer(logContext)) {
+ ShareCompletedFetch fetchForNode0 =
completedFetchWithAcquiredRecords(topicAPartition0, 0, 5);
+ ShareInFlightBatch<String, String> batch =
consumeFetchWithAcquiredRecords(fetchForNode0);
+ fetchBuffer.setNextInLineFetch(fetchForNode0);
+
+ // Replacing the next-in-line fetch with one for a different node
still retains the node 0 fetch because
+ // its acknowledgements are outstanding.
+
fetchBuffer.setNextInLineFetch(completedFetchWithAcquiredRecords(topicAPartition1,
1, 5));
+ assertEquals(partitions(topicAPartition0, topicAPartition1),
fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(0, 1), fetchBuffer.bufferedNodes());
+
+ // Once node 0's acknowledgements are taken, the retained fetch is
pruned.
+ batch.acknowledgeAll(AcknowledgeType.ACCEPT);
+ batch.takeAcknowledgedRecords();
+ assertEquals(partitions(topicAPartition1),
fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(1), fetchBuffer.bufferedNodes());
+ }
+ }
+
+ @Test
+ public void testRetainsEvictedFetchWithPendingRenewAcknowledgements() {
+ try (ShareFetchBuffer fetchBuffer = new ShareFetchBuffer(logContext)) {
+ ShareCompletedFetch completedFetch =
completedFetchWithAcquiredRecords(topicAPartition0, 0, 5);
+ ShareInFlightBatch<String, String> batch =
consumeFetchWithAcquiredRecords(completedFetch);
+ fetchBuffer.setNextInLineFetch(completedFetch);
+
+ // Renewing the records moves them out of the in-flight set, but
they are still held and the fetch remains buffered.
+ batch.acknowledgeAll(AcknowledgeType.RENEW);
+ Acknowledgements renewAcknowledgements =
batch.takeAcknowledgedRecords();
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+ assertEquals(partitions(topicAPartition0),
fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(0), fetchBuffer.bufferedNodes());
+
+ // The renewal completed and the records move to the renewed
state, and they are still held.
+ renewAcknowledgements.complete(null);
+ batch.renew(renewAcknowledgements);
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+ assertEquals(Set.of(0), fetchBuffer.bufferedNodes());
+
+ // The renewed records return to the in-flight set, so still
buffered.
+ batch.takeRenewals();
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+ assertEquals(Set.of(0), fetchBuffer.bufferedNodes());
+
+ // Finally the records accepted and the fetch is no longer
buffered.
+ batch.acknowledgeAll(AcknowledgeType.ACCEPT);
+ batch.takeAcknowledgedRecords();
+ assertFalse(completedFetch.hasPendingAcknowledgements());
+ assertEquals(partitions(), fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(), fetchBuffer.bufferedNodes());
+ }
+ }
+
+ @Test
+ public void testCloseClearsPendingAcknowledgementFetches() {
+ ShareFetchBuffer fetchBuffer = new ShareFetchBuffer(logContext);
+ try {
+ ShareCompletedFetch fetchForNode0 =
completedFetchWithAcquiredRecords(topicAPartition0, 0, 5);
+ consumeFetchWithAcquiredRecords(fetchForNode0);
+ fetchBuffer.setNextInLineFetch(fetchForNode0);
+
+ // Replace the next-in-line fetch with one for a different node
+
fetchBuffer.setNextInLineFetch(completedFetchWithAcquiredRecords(topicAPartition1,
1, 5));
+ assertEquals(Set.of(0, 1), fetchBuffer.bufferedNodes());
+ } finally {
+ fetchBuffer.close();
+ }
+
+ assertEquals(partitions(), fetchBuffer.bufferedPartitions());
+ assertEquals(Set.of(), fetchBuffer.bufferedNodes());
+ }
+
@Test
public void testWakeup() throws Exception {
try (ShareFetchBuffer fetchBuffer = new ShareFetchBuffer(logContext)) {
@@ -162,14 +274,31 @@ public class ShareFetchBufferTest {
}
}
+ private ShareInFlightBatch<String, String>
consumeFetchWithAcquiredRecords(ShareCompletedFetch completedFetch) {
+ ShareInFlightBatch<String, String> batch =
completedFetch.fetchRecords(deserializers, 100, false);
+ assertTrue(completedFetch.isConsumed());
+ assertTrue(completedFetch.hasPendingAcknowledgements());
+ return batch;
+ }
+
private ShareCompletedFetch completedFetch(TopicIdPartition tp) {
+ return completedFetch(tp, 0, new
ShareFetchResponseData.PartitionData());
+ }
+
+ private ShareCompletedFetch
completedFetchWithAcquiredRecords(TopicIdPartition tp, int nodeId, int
numRecords) {
+ ShareFetchResponseData.PartitionData partitionData = new
ShareFetchResponseData.PartitionData()
+ .setRecords(newRecords(0, numRecords))
+ .setAcquiredRecords(acquiredRecords(0, numRecords));
+ return completedFetch(tp, nodeId, partitionData);
+ }
+
+ private ShareCompletedFetch completedFetch(TopicIdPartition tp, int
nodeId, ShareFetchResponseData.PartitionData partitionData) {
ShareFetchMetricsAggregator shareFetchMetricsAggregator = new
ShareFetchMetricsAggregator(shareFetchMetricsManager,
allPartitions.stream().map(TopicIdPartition::topicPartition).collect(Collectors.toSet()));
- ShareFetchResponseData.PartitionData partitionData = new
ShareFetchResponseData.PartitionData();
return new ShareCompletedFetch(
logContext,
BufferSupplier.create(),
- 0,
+ nodeId,
tp,
partitionData,
DEFAULT_ACQUISITION_LOCK_TIMEOUT_MS,
@@ -177,6 +306,22 @@ public class ShareFetchBufferTest {
ApiKeys.SHARE_FETCH.latestVersion());
}
+ private static Records newRecords(long baseOffset, int numRecords) {
+ try (MemoryRecordsBuilder builder =
MemoryRecords.builder(ByteBuffer.allocate(1024), Compression.NONE,
TimestampType.CREATE_TIME, baseOffset)) {
+ for (int i = 0; i < numRecords; i++) {
+ builder.append(0L, "key".getBytes(), "value".getBytes());
+ }
+ return builder.build();
+ }
+ }
+
+ private static List<ShareFetchResponseData.AcquiredRecords>
acquiredRecords(long firstOffset, int numRecords) {
+ return List.of(new ShareFetchResponseData.AcquiredRecords()
+ .setFirstOffset(firstOffset)
+ .setLastOffset(firstOffset + numRecords - 1)
+ .setDeliveryCount((short) 1));
+ }
+
/**
* This is a handy utility method for returning a set from a varargs array.
*/
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandlerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandlerTest.java
index 405adff250e..de997920a2e 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandlerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandlerTest.java
@@ -44,8 +44,10 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
+import static
org.apache.kafka.common.requests.ShareRequestMetadata.FINAL_EPOCH;
import static
org.apache.kafka.common.requests.ShareRequestMetadata.INITIAL_EPOCH;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -619,6 +621,44 @@ public class ShareSessionHandlerTest {
assertEquals(1, barPartition.acknowledgementBatches().size());
}
+ @ParameterizedTest
+ @MethodSource("shareFetchConfigProvider")
+ public void
testCloseEmptySessionBuildsFinalEpochShareFetch(ShareFetchConfig
shareFetchConfig) {
+ String groupId = "G1";
+ Uuid memberId = Uuid.randomUuid();
+ ShareSessionHandler handler = new ShareSessionHandler(LOG_CONTEXT, 1,
memberId);
+
+ Map<Uuid, String> topicNames = new HashMap<>();
+ Uuid fooId = addTopicId(topicNames, "foo");
+ TopicIdPartition foo0 = new TopicIdPartition(fooId, 0, "foo");
+
+ // Establish the session with foo0.
+ handler.addPartitionToFetch(foo0, null);
+ assertNotNull(handler.newShareFetchBuilder(groupId, shareFetchConfig,
false));
+ assertFalse(handler.isSessionEmpty());
+ ShareFetchResponse resp = ShareFetchResponse.of(Errors.NONE,
+ 0,
+ buildResponseData(new RespEntry("foo", 0, fooId)),
+ List.of(),
+ 0);
+ handler.handleResponse(resp, ApiKeys.SHARE_FETCH.latestVersion());
+
+ // Remove the only partition by building without adding the partition,
emptying it.
+ assertNotNull(handler.newShareFetchBuilder(groupId, shareFetchConfig,
false));
+ handler.handleResponse(ShareFetchResponse.of(Errors.NONE, 0, new
LinkedHashMap<>(), List.of(), 0), ApiKeys.SHARE_FETCH.latestVersion());
+ assertTrue(handler.isSessionEmpty());
+
+ // Closing the empty session builds a ShareFetch request with the
final epoch to close it on the broker.
+ handler.notifyClose();
+ ShareFetchRequest.Builder builder =
handler.newShareFetchBuilder(groupId, shareFetchConfig, true);
+ assertNotNull(builder);
+ ShareFetchRequestData requestData = builder.build().data();
+ assertEquals(memberId.toString(), requestData.memberId());
+ assertEquals(FINAL_EPOCH, requestData.shareSessionEpoch());
+ assertTrue(requestData.topics().isEmpty());
+ assertTrue(requestData.forgottenTopicsData().isEmpty());
+ }
+
private Uuid addTopicId(Map<Uuid, String> topicNames, String name) {
Uuid id = Uuid.randomUuid();
topicNames.put(id, name);
