diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java index 4c4d6353eff2f..16c76afe3f72a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java +++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java @@ -18,6 +18,7 @@ package org.apache.kafka.clients; import org.apache.kafka.common.Node; +import org.apache.kafka.common.errors.AuthenticationException; import org.apache.kafka.common.errors.DisconnectException; import org.apache.kafka.common.utils.Time; @@ -114,4 +115,21 @@ public static ClientResponse sendAndReceive(KafkaClient client, ClientRequest re } } + + /** + * Check if the code is disconnected and unavailable for immediate reconnection (i.e. if it is in + * reconnect backoff window following the disconnect). + */ + public static boolean isUnavailable(KafkaClient client, Node node, Time time) { + return client.connectionFailed(node) && client.connectionDelay(node, time.milliseconds()) > 0; + } + + /** + * Check for an authentication error on a given node and raise the exception if there is one. + */ + public static void maybeThrowAuthFailure(KafkaClient client, Node node) { + AuthenticationException exception = client.authenticationException(node); + if (exception != null) + throw exception; + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java index 9ab9a3b476307..b98f7989c4142 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java @@ -639,9 +639,9 @@ private void maybeOverrideClientId(Map configs) { } } - protected static Map appendDeserializerToConfig(Map configs, - Deserializer keyDeserializer, - Deserializer valueDeserializer) { + public static Map appendDeserializerToConfig(Map configs, + Deserializer keyDeserializer, + Deserializer valueDeserializer) { // validate deserializer configuration, if the passed deserializer instance is null, the user must explicitly set a valid deserializer configuration value Map newConfigs = new HashMap<>(configs); if (keyDeserializer != null) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java index 5ab804ccb7ae5..8d6390ca94da5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java @@ -274,11 +274,11 @@ public Fetch collectFetch() { try { while (recordsRemaining > 0) { - if (nextInLineFetch == null || nextInLineFetch.isConsumed) { + if (nextInLineFetch == null || nextInLineFetch.isConsumed()) { CompletedFetch records = completedFetches.peek(); if (records == null) break; - if (!records.initialized) { + if (!records.isInitialized()) { try { nextInLineFetch = initializeCompletedFetch(records); } catch (Exception e) { @@ -336,7 +336,7 @@ private Fetch fetchRecords(final int maxRecords) { throw new IllegalStateException("Missing position for fetchable partition " + nextInLineFetch.partition); } - if (nextInLineFetch.nextFetchOffset == position.offset) { + if (nextInLineFetch.nextFetchOffset() == position.offset) { List> partRecords = nextInLineFetch.fetchRecords(maxRecords); log.trace("Returning {} fetched records at offset {} for assigned partition {}", @@ -344,10 +344,10 @@ private Fetch fetchRecords(final int maxRecords) { boolean positionAdvanced = false; - if (nextInLineFetch.nextFetchOffset > position.offset) { + if (nextInLineFetch.nextFetchOffset() > position.offset) { SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition( - nextInLineFetch.nextFetchOffset, - nextInLineFetch.lastEpoch, + nextInLineFetch.nextFetchOffset(), + nextInLineFetch.lastEpoch(), position.currentLeader); log.trace("Updating fetch position from {} to {} for partition {} and returning {} records from `poll()`", position, nextPosition, nextInLineFetch.partition, partRecords.size()); @@ -369,7 +369,7 @@ private Fetch fetchRecords(final int maxRecords) { // these records aren't next in line based on the last consumed position, ignore them // they must be from an obsolete request log.debug("Ignoring fetched records for {} at offset {} since the current position is {}", - nextInLineFetch.partition, nextInLineFetch.nextFetchOffset, position); + nextInLineFetch.partition, nextInLineFetch.nextFetchOffset(), position); } } @@ -381,7 +381,7 @@ private Fetch fetchRecords(final int maxRecords) { private List fetchablePartitions() { Set exclude = new HashSet<>(); - if (nextInLineFetch != null && !nextInLineFetch.isConsumed) { + if (nextInLineFetch != null && !nextInLineFetch.isConsumed()) { exclude.add(nextInLineFetch.partition); } for (CompletedFetch completedFetch : completedFetches) { @@ -528,7 +528,7 @@ private CompletedFetch initializeCompletedFetch(final CompletedFetch private CompletedFetch handleInitializeCompletedFetchSuccess(final CompletedFetch completedFetch) { final TopicPartition tp = completedFetch.partition; - final long fetchOffset = completedFetch.nextFetchOffset; + final long fetchOffset = completedFetch.nextFetchOffset(); // we are interested in this fetch only if the beginning offset matches the // current consumed position @@ -586,14 +586,14 @@ private CompletedFetch handleInitializeCompletedFetchSuccess(final Complet }); } - completedFetch.initialized = true; + completedFetch.setInitialized(); return completedFetch; } private void handleInitializeCompletedFetchErrors(final CompletedFetch completedFetch, final Errors error) { final TopicPartition tp = completedFetch.partition; - final long fetchOffset = completedFetch.nextFetchOffset; + final long fetchOffset = completedFetch.nextFetchOffset(); if (error == Errors.NOT_LEADER_OR_FOLLOWER || error == Errors.REPLICA_NOT_AVAILABLE || diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java index 9d23b9a9c2473..83672fbc080ad 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java @@ -53,7 +53,8 @@ public class CommitRequestManager implements RequestManager { // TODO: current in ConsumerConfig but inaccessible in the internal package. private static final String THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED = "internal.throw.on.fetch.stable.offset.unsupported"; // TODO: We will need to refactor the subscriptionState - private final SubscriptionState subscriptionState; + private final SubscriptionState subscriptions; + private final LogContext logContext; private final Logger log; private final Optional autoCommitState; private final CoordinatorRequestManager coordinatorRequestManager; @@ -66,11 +67,12 @@ public class CommitRequestManager implements RequestManager { public CommitRequestManager( final Time time, final LogContext logContext, - final SubscriptionState subscriptionState, + final SubscriptionState subscriptions, final ConsumerConfig config, final CoordinatorRequestManager coordinatorRequestManager, final GroupState groupState) { Objects.requireNonNull(coordinatorRequestManager, "Coordinator is needed upon committing offsets"); + this.logContext = logContext; this.log = logContext.logger(getClass()); this.pendingRequests = new PendingRequests(); if (config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) { @@ -82,7 +84,7 @@ public CommitRequestManager( } this.coordinatorRequestManager = coordinatorRequestManager; this.groupState = groupState; - this.subscriptionState = subscriptionState; + this.subscriptions = subscriptions; this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG); this.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG); this.throwOnFetchStableOffsetUnsupported = config.getBoolean(THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED); @@ -99,7 +101,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList()); } - maybeAutoCommit(this.subscriptionState.allConsumed()); + maybeAutoCommit(this.subscriptions.allConsumed()); if (!pendingRequests.hasUnsentRequests()) { return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList()); } @@ -167,9 +169,9 @@ CompletableFuture sendAutoCommit(final Map { if (t instanceof RetriableCommitFailedException) { - log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, t); + log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, t.getMessage()); } else { - log.warn("Asynchronous auto-commit of offsets {} failed: {}", allConsumedOffsets, t.getMessage()); + log.warn("Asynchronous auto-commit of offsets {} failed", allConsumedOffsets, t); } return null; }); @@ -241,7 +243,7 @@ public OffsetFetchRequestState(final Set partitions, final GroupState.Generation generation, final long retryBackoffMs, final long retryBackoffMaxMs) { - super(retryBackoffMs, retryBackoffMaxMs); + super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, retryBackoffMaxMs); this.requestedPartitions = partitions; this.requestedGeneration = generation; this.future = new CompletableFuture<>(); @@ -366,6 +368,16 @@ private CompletableFuture> chainFuture(fi } }); } + + @Override + public String toString() { + return "OffsetFetchRequestState{" + + "requestedPartitions=" + requestedPartitions + + ", requestedGeneration=" + requestedGeneration + + ", future=" + future + + ", " + toStringBase() + + '}'; + } } /** diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java index 96319b18ff170..26a134c1b8d1f 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java @@ -57,17 +57,12 @@ * @param Record key type * @param Record value type */ -class CompletedFetch { +public class CompletedFetch { final TopicPartition partition; final FetchResponseData.PartitionData partitionData; final short requestVersion; - long nextFetchOffset; - Optional lastEpoch; - boolean isConsumed = false; - boolean initialized = false; - private final Logger log; private final SubscriptionState subscriptions; private final FetchConfig fetchConfig; @@ -84,6 +79,10 @@ class CompletedFetch { private CloseableIterator records; private Exception cachedRecordException = null; private boolean corruptLastRecord = false; + private long nextFetchOffset; + private Optional lastEpoch; + private boolean isConsumed = false; + private boolean initialized = false; CompletedFetch(LogContext logContext, SubscriptionState subscriptions, @@ -109,6 +108,27 @@ class CompletedFetch { this.abortedTransactions = abortedTransactions(partitionData); } + long nextFetchOffset() { + return nextFetchOffset; + } + + Optional lastEpoch() { + return lastEpoch; + } + + boolean isInitialized() { + return initialized; + } + + void setInitialized() { + this.initialized = true; + } + + public boolean isConsumed() { + return isConsumed; + } + + /** * After each partition is parsed, we update the current metric totals with the total bytes * and number of records parsed. After all partitions have reported, we write the metric. diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java index 9a119a18b331c..97c64f93750cd 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java @@ -20,6 +20,7 @@ import org.apache.kafka.clients.ClientResponse; import org.apache.kafka.clients.KafkaClient; import org.apache.kafka.clients.Metadata; +import org.apache.kafka.clients.NetworkClientUtils; import org.apache.kafka.clients.RequestCompletionHandler; import org.apache.kafka.common.Node; import org.apache.kafka.common.errors.AuthenticationException; @@ -558,7 +559,7 @@ public void close() throws IOException { public boolean isUnavailable(Node node) { lock.lock(); try { - return client.connectionFailed(node) && client.connectionDelay(node, time.milliseconds()) > 0; + return NetworkClientUtils.isUnavailable(client, node, time); } finally { lock.unlock(); } @@ -570,9 +571,7 @@ public boolean isUnavailable(Node node) { public void maybeThrowAuthFailure(Node node) { lock.lock(); try { - AuthenticationException exception = client.authenticationException(node); - if (exception != null) - throw exception; + NetworkClientUtils.maybeThrowAuthFailure(client, node); } finally { lock.unlock(); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java index bc45c58a36dcc..241760d4c22dd 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java @@ -34,13 +34,13 @@ /** * This is responsible for timing to send the next {@link FindCoordinatorRequest} based on the following criteria: - * + *

* Whether there is an existing coordinator. * Whether there is an inflight request. * Whether the backoff timer has expired. * The {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} contains either a wait timer * or a singleton list of {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest}. - * + *

* The {@link FindCoordinatorRequest} will be handled by the {@link #onResponse(long, FindCoordinatorResponse)} callback, which * subsequently invokes {@code onResponse} to handle the exception and response. Note that the coordinator node will be * marked {@code null} upon receiving a failure. @@ -70,7 +70,12 @@ public CoordinatorRequestManager( this.log = logContext.logger(this.getClass()); this.nonRetriableErrorHandler = errorHandler; this.groupId = groupId; - this.coordinatorRequestState = new RequestState(retryBackoffMs, retryBackoffMaxMs); + this.coordinatorRequestState = new RequestState( + logContext, + CoordinatorRequestManager.class.getSimpleName(), + retryBackoffMs, + retryBackoffMaxMs + ); } /** @@ -218,5 +223,4 @@ private void onResponse( public Optional coordinator() { return Optional.ofNullable(this.coordinator); } - } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java index d0656daeb6635..449d5a471cc9a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java @@ -17,18 +17,27 @@ package org.apache.kafka.clients.consumer.internals; import org.apache.kafka.common.utils.ExponentialBackoff; +import org.apache.kafka.common.utils.LogContext; +import org.slf4j.Logger; class RequestState { + + private final Logger log; + protected final String owner; final static int RETRY_BACKOFF_EXP_BASE = 2; final static double RETRY_BACKOFF_JITTER = 0.2; - private final ExponentialBackoff exponentialBackoff; - private long lastSentMs = -1; - private long lastReceivedMs = -1; - private int numAttempts = 0; - private long backoffMs = 0; + protected final ExponentialBackoff exponentialBackoff; + protected long lastSentMs = -1; + protected long lastReceivedMs = -1; + protected int numAttempts = 0; + protected long backoffMs = 0; - public RequestState(final long retryBackoffMs, + public RequestState(final LogContext logContext, + final String owner, + final long retryBackoffMs, final long retryBackoffMaxMs) { + this.log = logContext.logger(RequestState.class); + this.owner = owner; this.exponentialBackoff = new ExponentialBackoff( retryBackoffMs, RETRY_BACKOFF_EXP_BASE, @@ -37,10 +46,14 @@ public RequestState(final long retryBackoffMs, } // Visible for testing - RequestState(final long retryBackoffMs, + RequestState(final LogContext logContext, + final String owner, + final long retryBackoffMs, final int retryBackoffExpBase, final long retryBackoffMaxMs, final double jitter) { + this.log = logContext.logger(RequestState.class); + this.owner = owner; this.exponentialBackoff = new ExponentialBackoff( retryBackoffMs, retryBackoffExpBase, @@ -65,13 +78,19 @@ public boolean canSendRequest(final long currentTimeMs) { return true; } - if (this.lastReceivedMs == -1 || - this.lastReceivedMs < this.lastSentMs) { - // there is an inflight request + if (this.lastReceivedMs == -1 || this.lastReceivedMs < this.lastSentMs) { + log.trace("An inflight request already exists for {}", this); return false; } - return requestBackoffExpired(currentTimeMs); + long remainingBackoffMs = remainingBackoffMs(currentTimeMs); + + if (remainingBackoffMs <= 0) { + return true; + } else { + log.trace("{} ms remain before another request should be sent for {}", remainingBackoffMs, this); + return false; + } } public void onSendAttempt(final long currentTimeMs) { @@ -105,12 +124,29 @@ public void onFailedAttempt(final long currentTimeMs) { this.numAttempts++; } - private boolean requestBackoffExpired(final long currentTimeMs) { - return remainingBackoffMs(currentTimeMs) <= 0; - } - long remainingBackoffMs(final long currentTimeMs) { long timeSinceLastReceiveMs = currentTimeMs - this.lastReceivedMs; return Math.max(0, backoffMs - timeSinceLastReceiveMs); } -} + + /** + * This method appends the instance variables together in a simple String of comma-separated key value pairs. + * This allows subclasses to include these values and not have to duplicate each variable, helping to prevent + * any variables from being omitted when new ones are added. + * + * @return String version of instance variables. + */ + protected String toStringBase() { + return "owner='" + owner + '\'' + + ", exponentialBackoff=" + exponentialBackoff + + ", lastSentMs=" + lastSentMs + + ", lastReceivedMs=" + lastReceivedMs + + ", numAttempts=" + numAttempts + + ", backoffMs=" + backoffMs; + } + + @Override + public String toString() { + return "RequestState{" + toStringBase() + '}'; + } +} \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java b/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java index 1c4482660a9b6..15e0cf2ba4b33 100644 --- a/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java +++ b/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java @@ -58,4 +58,14 @@ public long backoff(long attempts) { long backoffValue = (long) (randomFactor * term); return backoffValue > maxInterval ? maxInterval : backoffValue; } + + @Override + public String toString() { + return "ExponentialBackoff{" + + "multiplier=" + multiplier + + ", expMax=" + expMax + + ", initialInterval=" + initialInterval + + ", jitter=" + jitter + + '}'; + } } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java index 344df5df92b12..365fb1a7e6a8d 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.clients.consumer.internals; +import org.apache.kafka.common.utils.LogContext; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -25,6 +26,8 @@ public class RequestStateTest { @Test public void testRequestStateSimple() { RequestState state = new RequestState( + new LogContext(), + this.getClass().getSimpleName(), 100, 2, 1000,