collectFetch() {
- return fetchCollector.collectFetch(fetchBuffer);
+ return requestFutures;
}
}
\ No newline at end of file
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
index 2e2ed9f2ab461..2172313054d10 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
@@ -18,6 +18,8 @@
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
+import org.apache.kafka.clients.consumer.internals.events.ErrorBackgroundEvent;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
@@ -58,6 +60,7 @@
* See {@link HeartbeatRequestState} for more details.
*/
public class HeartbeatRequestManager implements RequestManager {
+
private final Logger logger;
private final Time time;
@@ -90,22 +93,22 @@ public class HeartbeatRequestManager implements RequestManager {
/**
* ErrorEventHandler allows the background thread to propagate errors back to the user
*/
- private final ErrorEventHandler nonRetriableErrorHandler;
+ private final BackgroundEventHandler backgroundEventHandler;
public HeartbeatRequestManager(
- final Time time,
final LogContext logContext,
+ final Time time,
final ConsumerConfig config,
final CoordinatorRequestManager coordinatorRequestManager,
final SubscriptionState subscriptions,
final MembershipManager membershipManager,
- final ErrorEventHandler nonRetriableErrorHandler) {
+ final BackgroundEventHandler backgroundEventHandler) {
this.coordinatorRequestManager = coordinatorRequestManager;
this.time = time;
this.logger = logContext.logger(getClass());
this.subscriptions = subscriptions;
this.membershipManager = membershipManager;
- this.nonRetriableErrorHandler = nonRetriableErrorHandler;
+ this.backgroundEventHandler = backgroundEventHandler;
this.rebalanceTimeoutMs = config.getInt(CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG);
long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
long retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
@@ -122,7 +125,7 @@ public HeartbeatRequestManager(
final SubscriptionState subscriptions,
final MembershipManager membershipManager,
final HeartbeatRequestState heartbeatRequestState,
- final ErrorEventHandler nonRetriableErrorHandler) {
+ final BackgroundEventHandler backgroundEventHandler) {
this.logger = logContext.logger(this.getClass());
this.time = time;
this.subscriptions = subscriptions;
@@ -130,7 +133,7 @@ public HeartbeatRequestManager(
this.coordinatorRequestManager = coordinatorRequestManager;
this.heartbeatRequestState = heartbeatRequestState;
this.membershipManager = membershipManager;
- this.nonRetriableErrorHandler = nonRetriableErrorHandler;
+ this.backgroundEventHandler = backgroundEventHandler;
}
/**
@@ -146,18 +149,14 @@ public HeartbeatRequestManager(
*/
@Override
public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
- if (!coordinatorRequestManager.coordinator().isPresent() || !membershipManager.shouldSendHeartbeat()) {
- return new NetworkClientDelegate.PollResult(
- Long.MAX_VALUE, Collections.emptyList());
- }
+ if (!coordinatorRequestManager.coordinator().isPresent() || !membershipManager.shouldSendHeartbeat())
+ return NetworkClientDelegate.PollResult.EMPTY;
// TODO: We will need to send a heartbeat response after partitions being revoke. This needs to be
// implemented either with or after the partition reconciliation logic.
- if (!heartbeatRequestState.canSendRequest(currentTimeMs)) {
- return new NetworkClientDelegate.PollResult(
- heartbeatRequestState.nextHeartbeatMs(currentTimeMs),
- Collections.emptyList());
- }
+ if (!heartbeatRequestState.canSendRequest(currentTimeMs))
+ return new NetworkClientDelegate.PollResult(heartbeatRequestState.nextHeartbeatMs(currentTimeMs));
+
this.heartbeatRequestState.onSendAttempt(currentTimeMs);
NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest();
return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
@@ -198,7 +197,7 @@ private void onFailure(final Throwable exception, final long responseTimeMs) {
this.heartbeatRequestState.onFailedAttempt(responseTimeMs);
if (exception instanceof RetriableException) {
String message = String.format("GroupHeartbeatRequest failed because of the retriable exception. " +
- "Will retry in %s ms: {}",
+ "Will retry in %s ms: %s",
heartbeatRequestState.remainingBackoffMs(responseTimeMs),
exception.getMessage());
logger.debug(message);
@@ -223,12 +222,13 @@ private void onErrorResponse(final ConsumerGroupHeartbeatResponse response,
final long currentTimeMs) {
Errors error = Errors.forCode(response.data().errorCode());
String errorMessage = response.data().errorMessage();
+ String message;
// TODO: upon encountering a fatal/fenced error, trigger onPartitionLost logic to give up the current
// assignments.
switch (error) {
case NOT_COORDINATOR:
// the manager should retry immediately when the coordinator node becomes available again
- String message = String.format("GroupHeartbeatRequest failed because the group coordinator %s is incorrect. " +
+ message = String.format("GroupHeartbeatRequest failed because the group coordinator %s is incorrect. " +
"Will attempt to find the coordinator again and retry",
coordinatorRequestManager.coordinator());
logInfo(message, response, currentTimeMs);
@@ -274,7 +274,7 @@ private void onErrorResponse(final ConsumerGroupHeartbeatResponse response,
break;
case FENCED_MEMBER_EPOCH:
- message = String.format("GroupHeartbeatRequest failed because member epoch %s is invalid. " +
+ message = String.format("GroupHeartbeatRequest failed because member ID %s with epoch %s is invalid. " +
"Will abandon all partitions and rejoin the group",
membershipManager.memberId(), membershipManager.memberEpoch());
logInfo(message, response, currentTimeMs);
@@ -282,7 +282,7 @@ private void onErrorResponse(final ConsumerGroupHeartbeatResponse response,
break;
case UNKNOWN_MEMBER_ID:
- message = String.format("GroupHeartbeatRequest failed because member id %s is invalid. " +
+ message = String.format("GroupHeartbeatRequest failed because member of unknown ID %s with epoch %s is invalid. " +
"Will abandon all partitions and rejoin the group",
membershipManager.memberId(), membershipManager.memberEpoch());
logInfo(message, response, currentTimeMs);
@@ -307,7 +307,7 @@ private void logInfo(final String message,
}
private void handleFatalFailure(Throwable error) {
- nonRetriableErrorHandler.handle(error);
+ backgroundEventHandler.add(new ErrorBackgroundEvent(error));
membershipManager.transitionToFailed();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
index 2e61b0788f8c0..47485ed46ca57 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
@@ -16,8 +16,10 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.ClientRequest;
import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.clients.ClientUtils;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.NetworkClientUtils;
import org.apache.kafka.clients.RequestCompletionHandler;
@@ -26,6 +28,7 @@
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
@@ -35,27 +38,29 @@
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Collections;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
-import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_MAX_INFLIGHT_REQUESTS_PER_CONNECTION;
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
/**
* A wrapper around the {@link org.apache.kafka.clients.NetworkClient} to handle network poll and send operations.
*/
public class NetworkClientDelegate implements AutoCloseable {
+
private final KafkaClient client;
private final Time time;
private final Logger log;
private final int requestTimeoutMs;
private final Queue unsentRequests;
private final long retryBackoffMs;
- private final Set tryConnectNodes;
public NetworkClientDelegate(
final Time time,
@@ -68,9 +73,40 @@ public NetworkClientDelegate(
this.unsentRequests = new ArrayDeque<>();
this.requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
- this.tryConnectNodes = new HashSet<>();
}
+ // Visible for testing
+ Queue unsentRequests() {
+ return unsentRequests;
+ }
+
+ /**
+ * Check if the node is disconnected and unavailable for immediate reconnection (i.e. if it is in
+ * reconnect backoff window following the disconnect).
+ *
+ * @param node {@link Node} to check for availability
+ * @see NetworkClientUtils#isUnavailable(KafkaClient, Node, Time)
+ */
+ public boolean isUnavailable(Node node) {
+ return NetworkClientUtils.isUnavailable(client, node, time);
+ }
+
+ /**
+ * Checks for an authentication error on a given node and throws the exception if it exists.
+ *
+ * @param node {@link Node} to check for a previous {@link AuthenticationException}; if found it is thrown
+ * @see NetworkClientUtils#maybeThrowAuthFailure(KafkaClient, Node)
+ */
+ public void maybeThrowAuthFailure(Node node) {
+ NetworkClientUtils.maybeThrowAuthFailure(client, node);
+ }
+
+ /**
+ * Initiate a connection if currently possible. This is only really useful for resetting
+ * the failed status of a socket.
+ *
+ * @param node The node to connect to
+ */
public void tryConnect(Node node) {
NetworkClientUtils.tryConnect(client, node, time);
}
@@ -81,7 +117,6 @@ public void tryConnect(Node node) {
*
* @param timeoutMs timeout time
* @param currentTimeMs current time
- * @return a list of client response
*/
public void poll(final long timeoutMs, final long currentTimeMs) {
trySend(currentTimeMs);
@@ -119,8 +154,7 @@ private void trySend(final long currentTimeMs) {
}
}
- private boolean doSend(final UnsentRequest r,
- final long currentTimeMs) {
+ boolean doSend(final UnsentRequest r, final long currentTimeMs) {
Node node = r.node.orElse(client.leastLoadedNode(currentTimeMs));
if (node == null || nodeUnavailable(node)) {
log.debug("No broker available to send the request: {}. Retrying.", r);
@@ -137,7 +171,7 @@ private boolean doSend(final UnsentRequest r,
return true;
}
- private void checkDisconnects(final long currentTimeMs) {
+ protected void checkDisconnects(final long currentTimeMs) {
// Check the connection of the unsent request. Disconnect the disconnected node if it is unable to be connected.
Iterator iter = unsentRequests.iterator();
while (iter.hasNext()) {
@@ -190,26 +224,47 @@ public void close() throws IOException {
this.client.close();
}
+ public long addAll(PollResult pollResult) {
+ addAll(pollResult.unsentRequests);
+ return pollResult.timeUntilNextPollMs;
+ }
+
public void addAll(final List requests) {
- requests.forEach(u -> {
- u.setTimer(this.time, this.requestTimeoutMs);
- });
- this.unsentRequests.addAll(requests);
+ if (!requests.isEmpty()) {
+ requests.forEach(ur -> ur.setTimer(time, requestTimeoutMs));
+ unsentRequests.addAll(requests);
+ }
}
public static class PollResult {
+
+ public static final long WAIT_FOREVER = Long.MAX_VALUE;
+ public static final PollResult EMPTY = new PollResult(WAIT_FOREVER);
public final long timeUntilNextPollMs;
public final List unsentRequests;
- public PollResult(final long timeMsTillNextPoll, final List unsentRequests) {
- this.timeUntilNextPollMs = timeMsTillNextPoll;
+ public PollResult(final long timeUntilNextPollMs, final List unsentRequests) {
+ this.timeUntilNextPollMs = timeUntilNextPollMs;
this.unsentRequests = Collections.unmodifiableList(unsentRequests);
}
+
+ public PollResult(final List unsentRequests) {
+ this(WAIT_FOREVER, unsentRequests);
+ }
+
+ public PollResult(final UnsentRequest unsentRequest) {
+ this(Collections.singletonList(unsentRequest));
+ }
+
+ public PollResult(final long timeUntilNextPollMs) {
+ this(timeUntilNextPollMs, Collections.emptyList());
+ }
}
+
public static class UnsentRequest {
private final AbstractRequest.Builder> requestBuilder;
private final FutureCompletionHandler handler;
- private Optional node; // empty if random node can be chosen
+ private final Optional node; // empty if random node can be chosen
private Timer timer;
public UnsentRequest(final AbstractRequest.Builder> requestBuilder,
@@ -227,6 +282,12 @@ public UnsentRequest(final AbstractRequest.Builder> requestBuilder,
this.handler.future().whenComplete(callback);
}
+ public UnsentRequest(final AbstractRequest.Builder> requestBuilder,
+ final Node node,
+ final BiConsumer callback) {
+ this(requestBuilder, Optional.of(node), callback);
+ }
+
public void setTimer(final Time time, final long requestTimeoutMs) {
this.timer = time.timer(requestTimeoutMs);
}
@@ -243,6 +304,10 @@ AbstractRequest.Builder> requestBuilder() {
return requestBuilder;
}
+ Optional node() {
+ return node;
+ }
+
@Override
public String toString() {
return "UnsentRequest{" +
@@ -291,4 +356,32 @@ public CompletableFuture future() {
return future;
}
}
+
+ /**
+ * Creates a {@link Supplier} for deferred creation during invocation by
+ * {@link ConsumerNetworkThread}.
+ */
+ public static Supplier supplier(final Time time,
+ final LogContext logContext,
+ final ConsumerMetadata metadata,
+ final ConsumerConfig config,
+ final ApiVersions apiVersions,
+ final Metrics metrics,
+ final FetchMetricsManager fetchMetricsManager) {
+ return new CachedSupplier() {
+ @Override
+ protected NetworkClientDelegate create() {
+ KafkaClient client = ClientUtils.createNetworkClient(config,
+ metrics,
+ CONSUMER_METRIC_GROUP_PREFIX,
+ logContext,
+ apiVersions,
+ time,
+ CONSUMER_MAX_INFLIGHT_REQUESTS_PER_CONNECTION,
+ metadata,
+ fetchMetricsManager.throttleTimeSensor());
+ return new NetworkClientDelegate(time, config, logContext, client);
+ }
+ };
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java
index b7fdefeb0d100..9239811f7d602 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java
@@ -19,6 +19,7 @@
import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.consumer.LogTruncationException;
+import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
@@ -262,7 +263,7 @@ private Long offsetResetStrategyTimestamp(final TopicPartition partition) {
else if (strategy == OffsetResetStrategy.LATEST)
return ListOffsetsRequest.LATEST_TIMESTAMP;
else
- return null;
+ throw new NoOffsetForPartitionException(partition);
}
static Set topicsForPartitions(Collection partitions) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java
index a1cb8d54a9778..700e2ab6e173d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java
@@ -24,6 +24,8 @@
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.ListOffsetData;
import org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.ListOffsetResult;
+import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
+import org.apache.kafka.clients.consumer.internals.events.ErrorBackgroundEvent;
import org.apache.kafka.common.ClusterResource;
import org.apache.kafka.common.ClusterResourceListener;
import org.apache.kafka.common.IsolationLevel;
@@ -83,6 +85,7 @@ public class OffsetsRequestManager implements RequestManager, ClusterResourceLis
private final Time time;
private final ApiVersions apiVersions;
private final NetworkClientDelegate networkClientDelegate;
+ private final BackgroundEventHandler backgroundEventHandler;
@SuppressWarnings("this-escape")
public OffsetsRequestManager(final SubscriptionState subscriptionState,
@@ -93,6 +96,7 @@ public OffsetsRequestManager(final SubscriptionState subscriptionState,
final long requestTimeoutMs,
final ApiVersions apiVersions,
final NetworkClientDelegate networkClientDelegate,
+ final BackgroundEventHandler backgroundEventHandler,
final LogContext logContext) {
requireNonNull(subscriptionState);
requireNonNull(metadata);
@@ -100,6 +104,7 @@ public OffsetsRequestManager(final SubscriptionState subscriptionState,
requireNonNull(time);
requireNonNull(apiVersions);
requireNonNull(networkClientDelegate);
+ requireNonNull(backgroundEventHandler);
requireNonNull(logContext);
this.metadata = metadata;
@@ -112,11 +117,12 @@ public OffsetsRequestManager(final SubscriptionState subscriptionState,
this.requestTimeoutMs = requestTimeoutMs;
this.apiVersions = apiVersions;
this.networkClientDelegate = networkClientDelegate;
+ this.backgroundEventHandler = backgroundEventHandler;
this.offsetFetcherUtils = new OffsetFetcherUtils(logContext, metadata, subscriptionState,
time, retryBackoffMs, apiVersions);
// Register the cluster metadata update callback. Note this only relies on the
// requestsToRetry initialized above, and won't be invoked until all managers are
- // initialized and the background thread started.
+ // initialized and the network thread started.
this.metadata.addClusterUpdateListener(this);
}
@@ -127,10 +133,10 @@ public OffsetsRequestManager(final SubscriptionState subscriptionState,
*/
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
- NetworkClientDelegate.PollResult pollResult =
- new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>(requestsToSend));
- this.requestsToSend.clear();
- return pollResult;
+ // Copy the outgoing request list and clear it.
+ List unsentRequests = new ArrayList<>(requestsToSend);
+ requestsToSend.clear();
+ return new NetworkClientDelegate.PollResult(unsentRequests);
}
/**
@@ -188,7 +194,14 @@ public CompletableFuture