diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/CommitFailedException.java b/clients/src/main/java/org/apache/kafka/clients/consumer/CommitFailedException.java index cae5b2aad5d37..2040216cd02c9 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/CommitFailedException.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/CommitFailedException.java @@ -28,6 +28,10 @@ public class CommitFailedException extends KafkaException { private static final long serialVersionUID = 1L; + public CommitFailedException(final String message) { + super(message); + } + public CommitFailedException() { super("Commit cannot be completed since the group has already " + "rebalanced and assigned the partitions to another member. This means that the time " + diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 95825c22d1598..883a048e78ce4 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1231,9 +1231,12 @@ private ConsumerRecords poll(final Timer timer, final boolean includeMetad client.maybeTriggerWakeup(); if (includeMetadataInTimeout) { - if (!updateAssignmentMetadataIfNeeded(timer)) { - return ConsumerRecords.empty(); - } + // try to update assignment metadata BUT do not need to block on the timer, + // since even if we are 1) in the middle of a rebalance or 2) have partitions + // with unknown starting positions we may still want to return some data + // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms + // to never block on completing the rebalance procedure if there's any + updateAssignmentMetadataIfNeeded(time.timer(0L)); } else { while (!updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE))) { log.warn("Still waiting for metadata"); @@ -1307,12 +1310,6 @@ private Map>> pollForFetches(Timer tim }); timer.update(pollTimer.currentTimeMs()); - // after the long poll, we should check whether the group needs to rebalance - // prior to returning data so that the group can stabilize faster - if (coordinator != null && coordinator.rejoinNeededOrPending()) { - return Collections.emptyMap(); - } - return fetcher.fetchedRecords(); } @@ -1332,8 +1329,16 @@ private Map>> pollForFetches(Timer tim * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method. * * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. - * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, - * or if there is an active group with the same groupId which is using group management. + * This fatal error can only occur if you are using automatic group management with {@link #subscribe(Collection)}, + * or if there is an active group with the same group.id which is using group management. In such cases, + * when you are trying to commit to partitions that are no longer assigned to this consumer because the + * consumer is for example no longer part of the group this exception would be thrown. + * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance + * so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first + * complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards. + * NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, + * and also for those partitions that are still assigned their fetch positions may have changed too + * if more records are returned from the {@link #poll(Duration)} call. * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this * function is called * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while @@ -1360,7 +1365,7 @@ public void commitSync() { * every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API * should not be used. *

- * This is a synchronous commits and will block until either the commit succeeds, an unrecoverable error is + * This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is * encountered (in which case it is thrown to the caller), or the passed timeout expires. *

* Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)} @@ -1368,7 +1373,15 @@ public void commitSync() { * * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, - * or if there is an active group with the same groupId which is using group management. + * or if there is an active group with the same group.id which is using group management. In such cases, + * when you are trying to commit to partitions that are no longer assigned to this consumer because the + * consumer is for example no longer part of the group this exception would be thrown. + * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance + * so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first + * complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards. + * NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, + * and also for those partitions that are still assigned their fetch positions may have changed too + * if more records are returned from the {@link #poll(Duration)} call. * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this * function is called * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while @@ -1402,9 +1415,10 @@ public void commitSync(Duration timeout) { * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API * should not be used. The committed offset should be the next message your application will consume, - * i.e. lastProcessedMessageOffset + 1. + * i.e. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used, + * then the committed offsets must belong to the currently auto-assigned partitions. *

- * This is a synchronous commits and will block until either the commit succeeds or an unrecoverable error is + * This is a synchronous commit and will block until either the commit succeeds or an unrecoverable error is * encountered (in which case it is thrown to the caller), or the timeout specified by {@code default.api.timeout.ms} expires * (in which case a {@link org.apache.kafka.common.errors.TimeoutException} is thrown to the caller). *

@@ -1414,7 +1428,16 @@ public void commitSync(Duration timeout) { * @param offsets A map of offsets by partition with associated metadata * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, - * or if there is an active group with the same groupId which is using group management. + * or if there is an active group with the same group.id which is using group management. In such cases, + * when you are trying to commit to partitions that are no longer assigned to this consumer because the + * consumer is for example no longer part of the group this exception would be thrown. + * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance + * so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first + * complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards. + * NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, + * and also for those partitions that are still assigned their fetch positions may have changed too + * if more records are returned from the {@link #poll(Duration)} call, so when you retry committing + * you should consider updating the passed in {@code offset} parameter. * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this * function is called * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while @@ -1435,38 +1458,48 @@ public void commitSync(final Map offsets) { } /** - * Commit the specified offsets for the specified list of topics and partitions. - *

- * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every - * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API - * should not be used. The committed offset should be the next message your application will consume, - * i.e. lastProcessedMessageOffset + 1. - *

- * This is a synchronous commits and will block until either the commit succeeds, an unrecoverable error is - * encountered (in which case it is thrown to the caller), or the timeout expires. - *

- * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)} - * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method. - * - * @param offsets A map of offsets by partition with associated metadata - * @param timeout The maximum amount of time to await completion of the offset commit - * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. - * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, - * or if there is an active group with the same groupId which is using group management. - * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this - * function is called - * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while - * this function is called - * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details - * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the - * configured groupId. See the exception for more details - * @throws java.lang.IllegalArgumentException if the committed offset is negative - * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata - * is too large or if the topic does not exist). - * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before successful completion - * of the offset commit - * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer instance gets fenced by broker. - */ + * Commit the specified offsets for the specified list of topics and partitions. + *

+ * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every + * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API + * should not be used. The committed offset should be the next message your application will consume, + * i.e. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used, + * then the committed offsets must belong to the currently auto-assigned partitions. + *

+ * This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is + * encountered (in which case it is thrown to the caller), or the timeout expires. + *

+ * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)} + * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method. + * + * @param offsets A map of offsets by partition with associated metadata + * @param timeout The maximum amount of time to await completion of the offset commit + * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. + * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, + * or if there is an active group with the same group.id which is using group management. In such cases, + * when you are trying to commit to partitions that are no longer assigned to this consumer because the + * consumer is for example no longer part of the group this exception would be thrown. + * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance + * so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first + * complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards. + * NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, + * and also for those partitions that are still assigned their fetch positions may have changed too + * if more records are returned from the {@link #poll(Duration)} call, so when you retry committing + * you should consider updating the passed in {@code offset} parameter. + * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this + * function is called + * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while + * this function is called + * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details + * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the + * configured groupId. See the exception for more details + * @throws java.lang.IllegalArgumentException if the committed offset is negative + * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata + * is too large or if the topic does not exist). + * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before successful completion + * of the offset commit + * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer instance gets fenced by broker. + */ @Override public void commitSync(final Map offsets, final Duration timeout) { acquireAndEnsureOpen(); @@ -1521,7 +1554,8 @@ public void commitAsync(OffsetCommitCallback callback) { * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API * should not be used. The committed offset should be the next message your application will consume, - * i.e. lastProcessedMessageOffset + 1. + * i.e. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used, + * then the committed offsets must belong to the currently auto-assigned partitions. *

* This is an asynchronous call and will not block. Any errors encountered are either passed to the callback * (if provided) or discarded. diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java index 383c1c82bcec2..19e6d7dc33008 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java @@ -18,6 +18,7 @@ import org.apache.kafka.common.TopicPartition; +import java.time.Duration; import java.util.Collection; import java.util.Map; @@ -37,6 +38,9 @@ public interface OffsetCommitCallback { * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. * This can only occur if you are using automatic group management with {@link KafkaConsumer#subscribe(Collection)}, * or if there is an active group with the same groupId which is using group management. + * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the commit failed because + * it is in the middle of a rebalance. In such cases + * commit could be retried after the rebalance is completed with the {@link #poll(Duration)} call. * @throws org.apache.kafka.common.errors.WakeupException if {@link KafkaConsumer#wakeup()} is called before or while this * function is called * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/RetriableCommitFailedException.java b/clients/src/main/java/org/apache/kafka/clients/consumer/RetriableCommitFailedException.java index e719e7c525f30..f44dce6187cda 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/RetriableCommitFailedException.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/RetriableCommitFailedException.java @@ -22,12 +22,6 @@ public class RetriableCommitFailedException extends RetriableException { private static final long serialVersionUID = 1L; - public static RetriableCommitFailedException withUnderlyingMessage(String additionalMessage) { - return new RetriableCommitFailedException("Offset commit failed with a retriable exception. " + - "You should retry committing the latest consumed offsets. " + - "The underlying error was: " + additionalMessage); - } - public RetriableCommitFailedException(Throwable t) { super("Offset commit failed with a retriable exception. You should retry committing " + "the latest consumed offsets.", t); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java index 9155378fb257e..1bd6acaf89c25 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java @@ -111,7 +111,7 @@ public abstract class AbstractCoordinator implements Closeable { public static final String HEARTBEAT_THREAD_PREFIX = "kafka-coordinator-heartbeat-thread"; - private enum MemberState { + protected enum MemberState { UNJOINED, // the client is not part of a group REBALANCING, // the client has begun rebalancing STABLE, // the client has joined and is sending heartbeats @@ -130,11 +130,12 @@ private enum MemberState { private MemberState state = MemberState.UNJOINED; private HeartbeatThread heartbeatThread = null; private RequestFuture joinFuture = null; + private RequestFuture findCoordinatorFuture = null; + volatile private RuntimeException findCoordinatorException = null; private Generation generation = Generation.NO_GENERATION; private long lastRebalanceStartMs = -1L; private long lastRebalanceEndMs = -1L; - private RequestFuture findCoordinatorFuture = null; /** * Initialize the coordination manager. @@ -226,6 +227,11 @@ protected synchronized boolean ensureCoordinatorReady(final Timer timer) { return true; do { + if (findCoordinatorException != null && !(findCoordinatorException instanceof RetriableException)) { + final RuntimeException fatalException = findCoordinatorException; + findCoordinatorException = null; + throw fatalException; + } final RequestFuture future = lookupCoordinator(); client.poll(future, timer); @@ -258,8 +264,20 @@ protected synchronized RequestFuture lookupCoordinator() { if (node == null) { log.debug("No broker available to send FindCoordinator request"); return RequestFuture.noBrokersAvailable(); - } else + } else { findCoordinatorFuture = sendFindCoordinatorRequest(node); + // remember the exception even after the future is cleared so that + // it can still be thrown by the ensureCoordinatorReady caller + findCoordinatorFuture.addListener(new RequestFutureListener() { + @Override + public void onSuccess(Void value) {} // do nothing + + @Override + public void onFailure(RuntimeException e) { + findCoordinatorException = e; + } + }); + } } return findCoordinatorFuture; } @@ -436,9 +454,9 @@ boolean joinGroupIfNeeded(final Timer timer) { resetJoinGroupFuture(); final RuntimeException exception = future.exception(); if (exception instanceof UnknownMemberIdException || - exception instanceof RebalanceInProgressException || - exception instanceof IllegalGenerationException || - exception instanceof MemberIdRequiredException) + exception instanceof RebalanceInProgressException || + exception instanceof IllegalGenerationException || + exception instanceof MemberIdRequiredException) continue; else if (!future.isRetriable()) throw exception; @@ -842,6 +860,10 @@ protected synchronized Generation generationIfStable() { return generation; } + protected synchronized boolean rebalanceInProgress() { + return this.state == MemberState.REBALANCING; + } + protected synchronized String memberId() { return generation.memberId; } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index f3f2f46fd02fa..22bd65637b13d 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -36,6 +36,7 @@ import org.apache.kafka.common.errors.FencedInstanceIdException; import org.apache.kafka.common.errors.GroupAuthorizationException; import org.apache.kafka.common.errors.InterruptException; +import org.apache.kafka.common.errors.RebalanceInProgressException; import org.apache.kafka.common.errors.RetriableException; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.TopicAuthorizationException; @@ -438,7 +439,7 @@ public boolean poll(Timer timer) { invokeCompletedOffsetCommitCallbacks(); - if (subscriptions.partitionsAutoAssigned()) { + if (subscriptions.hasAutoAssignedPartitions()) { if (protocol == null) { throw new IllegalStateException("User configured " + ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " to empty while trying to subscribe for group protocol to auto assign partitions"); @@ -706,7 +707,7 @@ public void onLeavePrepare() { // we should reset assignment and trigger the callback before leaving group Set droppedPartitions = new HashSet<>(subscriptions.assignedPartitions()); - if (subscriptions.partitionsAutoAssigned() && !droppedPartitions.isEmpty()) { + if (subscriptions.hasAutoAssignedPartitions() && !droppedPartitions.isEmpty()) { final Exception e; if (generation() != Generation.NO_GENERATION) { e = invokePartitionsRevoked(droppedPartitions); @@ -727,7 +728,7 @@ public void onLeavePrepare() { */ @Override public boolean rejoinNeededOrPending() { - if (!subscriptions.partitionsAutoAssigned()) + if (!subscriptions.hasAutoAssignedPartitions()) return false; // we need to rejoin if we performed the assignment and metadata has changed; @@ -759,14 +760,24 @@ public boolean refreshCommittedOffsetsIfNeeded(Timer timer) { final TopicPartition tp = entry.getKey(); final OffsetAndMetadata offsetAndMetadata = entry.getValue(); if (offsetAndMetadata != null) { - final ConsumerMetadata.LeaderAndEpoch leaderAndEpoch = metadata.leaderAndEpoch(tp); - final SubscriptionState.FetchPosition position = new SubscriptionState.FetchPosition( - offsetAndMetadata.offset(), offsetAndMetadata.leaderEpoch(), - leaderAndEpoch); - - log.info("Setting offset for partition {} to the committed offset {}", tp, position); + // first update the epoch if necessary entry.getValue().leaderEpoch().ifPresent(epoch -> this.metadata.updateLastSeenEpochIfNewer(entry.getKey(), epoch)); - this.subscriptions.seekUnvalidated(tp, position); + + // it's possible that the partition is no longer assigned when the response is received, + // so we need to ignore seeking if that's the case + if (this.subscriptions.isAssigned(tp)) { + final ConsumerMetadata.LeaderAndEpoch leaderAndEpoch = metadata.leaderAndEpoch(tp); + final SubscriptionState.FetchPosition position = new SubscriptionState.FetchPosition( + offsetAndMetadata.offset(), offsetAndMetadata.leaderEpoch(), + leaderAndEpoch); + + this.subscriptions.seekUnvalidated(tp, position); + + log.info("Setting offset for partition {} to the committed offset {}", tp, position); + } else { + log.info("Ignoring the returned {} since its partition {} is no longer assigned", + offsetAndMetadata, tp); + } } } return true; @@ -986,7 +997,7 @@ private void doAutoCommitOffsetsAsync() { commitOffsetsAsync(allConsumedOffsets, (offsets, exception) -> { if (exception != null) { - if (exception instanceof RetriableException) { + if (exception instanceof RetriableCommitFailedException) { log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", offsets, exception); nextAutoCommitTimer.updateAndReset(rebalanceConfig.retryBackoffMs); @@ -1066,16 +1077,28 @@ private RequestFuture sendOffsetCommitRequest(final Map futu future.raise(error); return; } else if (error == Errors.REBALANCE_IN_PROGRESS) { - /* Consumer never tries to commit offset in between join-group and sync-group, + /* Consumer should not try to commit offset in between join-group and sync-group, * and hence on broker-side it is not expected to see a commit offset request * during CompletingRebalance phase; if it ever happens then broker would return - * this error. In this case we should just treat as a fatal CommitFailed exception. - * However, we do not need to reset generations and just request re-join, such that - * if the caller decides to proceed and poll, it would still try to proceed and re-join normally. + * this error to indicate that we are still in the middle of a rebalance. + * In this case we would throw a RebalanceInProgressException, + * request re-join but do not reset generations. If the callers decide to retry they + * can go ahead and call poll to finish up the rebalance first, and then try commit again. */ requestRejoin(); - future.raise(new CommitFailedException()); + future.raise(new RebalanceInProgressException("Offset commit cannot be completed since the " + + "consumer group is executing a rebalance at the moment. You can try completing the rebalance " + + "by calling poll() and then retry commit again")); return; } else if (error == Errors.UNKNOWN_MEMBER_ID || error == Errors.ILLEGAL_GENERATION) { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/KafkaConsumerMetrics.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/KafkaConsumerMetrics.java index aab313320d4bd..0e750abc99b68 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/KafkaConsumerMetrics.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/KafkaConsumerMetrics.java @@ -26,17 +26,16 @@ import java.util.concurrent.TimeUnit; public class KafkaConsumerMetrics implements AutoCloseable { - private final Metrics metrics; private final MetricName lastPollMetricName; private final Sensor timeBetweenPollSensor; private final Sensor pollIdleSensor; + private final Metrics metrics; private long lastPollMs; private long pollStartMs; private long timeSinceLastPollMs; public KafkaConsumerMetrics(Metrics metrics, String metricGrpPrefix) { this.metrics = metrics; - String metricGroupName = metricGrpPrefix + "-metrics"; Measurable lastPoll = (mConfig, now) -> { if (lastPollMs == 0L) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index bd05909269ce7..92b997048e5f2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -198,7 +198,7 @@ private boolean changeSubscription(Set topicsToSubscribe) { * @param topics The topics to add to the group subscription */ synchronized boolean groupSubscribe(Collection topics) { - if (!partitionsAutoAssigned()) + if (!hasAutoAssignedPartitions()) throw new IllegalStateException(SUBSCRIPTION_EXCEPTION_MESSAGE); groupSubscription = new HashSet<>(groupSubscription); return groupSubscription.addAll(topics); @@ -224,6 +224,7 @@ public synchronized boolean assignFromUser(Set partitions) { assignmentId++; + // update the subscribed topics Set manualSubscribedTopics = new HashSet<>(); Map partitionToState = new HashMap<>(); for (TopicPartition partition : partitions) { @@ -231,8 +232,10 @@ public synchronized boolean assignFromUser(Set partitions) { if (state == null) state = new TopicPartitionState(); partitionToState.put(partition, state); + manualSubscribedTopics.add(partition.topic()); } + this.assignment.set(partitionToState); return changeSubscription(manualSubscribedTopics); } @@ -267,7 +270,7 @@ public synchronized boolean checkAssignmentMatchedSubscription(Collection assignments) { - if (!this.partitionsAutoAssigned()) + if (!this.hasAutoAssignedPartitions()) throw new IllegalArgumentException("Attempt to dynamically assign partitions while manual assignment in use"); Map assignedPartitionStates = new HashMap<>(assignments.size()); @@ -322,7 +325,7 @@ synchronized boolean matchesSubscribedPattern(String topic) { } public synchronized Set subscription() { - if (partitionsAutoAssigned()) + if (hasAutoAssignedPartitions()) return this.subscription; return Collections.emptySet(); } @@ -416,7 +419,7 @@ synchronized List fetchablePartitions(Predicate .collect(Collectors.toList()); } - synchronized boolean partitionsAutoAssigned() { + public synchronized boolean hasAutoAssignedPartitions() { return this.subscriptionType == SubscriptionType.AUTO_TOPICS || this.subscriptionType == SubscriptionType.AUTO_PATTERN; } diff --git a/clients/src/test/java/org/apache/kafka/clients/MockClient.java b/clients/src/test/java/org/apache/kafka/clients/MockClient.java index f9fb96ce0c0a8..ca3a88d687098 100644 --- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java +++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java @@ -302,7 +302,6 @@ private long elapsedTimeMs(long currentTimeMs, long startTimeMs) { return Math.max(0, currentTimeMs - startTimeMs); } - private void checkTimeoutOfPendingRequests(long nowMs) { ClientRequest request = requests.peek(); while (request != null && elapsedTimeMs(nowMs, request.createdTimeMs()) > request.requestTimeoutMs()) { @@ -334,7 +333,6 @@ public void respond(RequestMatcher matcher, AbstractResponse response) { // Utility method to enable out of order responses public void respondToRequest(ClientRequest clientRequest, AbstractResponse response) { - AbstractRequest request = clientRequest.requestBuilder().build(); requests.remove(clientRequest); short version = clientRequest.requestBuilder().latestAllowedVersion(); responses.add(new ClientResponse(clientRequest.makeHeader(version), clientRequest.callback(), clientRequest.destination(), diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index 879eaa17432d9..c7ebf48cadc97 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -1773,6 +1773,120 @@ public void testRebalanceException() { assertTrue(subscription.assignedPartitions().isEmpty()); } + @Test + public void testReturnRecordsDuringRebalance() { + Time time = new MockTime(1L); + SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST); + ConsumerMetadata metadata = createMetadata(subscription); + MockClient client = new MockClient(time, metadata); + ConsumerPartitionAssignor assignor = new CooperativeStickyAssignor(); + KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); + + initMetadata(client, Utils.mkMap(Utils.mkEntry(topic, 1), Utils.mkEntry(topic2, 1), Utils.mkEntry(topic3, 1))); + + consumer.subscribe(Arrays.asList(topic, topic2), getConsumerRebalanceListener(consumer)); + Node node = metadata.fetch().nodes().get(0); + Node coordinator = prepareRebalance(client, node, assignor, Arrays.asList(tp0, t2p0), null); + + // a first rebalance to get the assignment, we need two poll calls since we need two round trips to finish join / sync-group + consumer.poll(Duration.ZERO); + consumer.poll(Duration.ZERO); + + assertEquals(Utils.mkSet(topic, topic2), consumer.subscription()); + assertEquals(Utils.mkSet(tp0, t2p0), consumer.assignment()); + + // prepare a response of the outstanding fetch so that we have data available on the next poll + Map fetches1 = new HashMap<>(); + fetches1.put(tp0, new FetchInfo(0, 1)); + fetches1.put(t2p0, new FetchInfo(0, 10)); + client.respondFrom(fetchResponse(fetches1), node); + + ConsumerRecords records = consumer.poll(Duration.ZERO); + + // verify that the fetch occurred as expected + assertEquals(11, records.count()); + assertEquals(1L, consumer.position(tp0)); + assertEquals(10L, consumer.position(t2p0)); + + // prepare the next response of the prefetch + fetches1.clear(); + fetches1.put(tp0, new FetchInfo(1, 1)); + fetches1.put(t2p0, new FetchInfo(10, 20)); + client.respondFrom(fetchResponse(fetches1), node); + + // subscription change + consumer.subscribe(Arrays.asList(topic, topic3), getConsumerRebalanceListener(consumer)); + + // verify that subscription has changed but assignment is still unchanged + assertEquals(Utils.mkSet(topic, topic3), consumer.subscription()); + assertEquals(Utils.mkSet(tp0, t2p0), consumer.assignment()); + + // mock the offset commit response for to be revoked partitions + Map partitionOffsets1 = new HashMap<>(); + partitionOffsets1.put(t2p0, 10L); + AtomicBoolean commitReceived = prepareOffsetCommitResponse(client, coordinator, partitionOffsets1); + + // poll once which would not complete the rebalance + records = consumer.poll(Duration.ZERO); + + // clear out the prefetch so it doesn't interfere with the rest of the test + fetches1.clear(); + fetches1.put(tp0, new FetchInfo(2, 1)); + client.respondFrom(fetchResponse(fetches1), node); + + // verify that the fetch still occurred as expected + assertEquals(Utils.mkSet(topic, topic3), consumer.subscription()); + assertEquals(Collections.singleton(tp0), consumer.assignment()); + assertEquals(1, records.count()); + assertEquals(2L, consumer.position(tp0)); + + // verify that the offset commits occurred as expected + assertTrue(commitReceived.get()); + + // mock rebalance responses + client.respondFrom(joinGroupFollowerResponse(assignor, 2, "memberId", "leaderId", Errors.NONE), coordinator); + client.prepareResponseFrom(syncGroupResponse(Arrays.asList(tp0, t3p0), Errors.NONE), coordinator); + + // we need to poll 1) for getting the join response, and then send the sync request; + // 2) for getting the sync response + records = consumer.poll(Duration.ZERO); + + // should not finish the response yet + assertEquals(Utils.mkSet(topic, topic3), consumer.subscription()); + assertEquals(Collections.singleton(tp0), consumer.assignment()); + assertEquals(1, records.count()); + assertEquals(3L, consumer.position(tp0)); + + fetches1.clear(); + fetches1.put(tp0, new FetchInfo(3, 1)); + client.respondFrom(fetchResponse(fetches1), node); + + records = consumer.poll(Duration.ZERO); + + // should have t3 but not sent yet the t3 records + assertEquals(Utils.mkSet(topic, topic3), consumer.subscription()); + assertEquals(Utils.mkSet(tp0, t3p0), consumer.assignment()); + assertEquals(1, records.count()); + assertEquals(4L, consumer.position(tp0)); + assertEquals(0L, consumer.position(t3p0)); + + fetches1.clear(); + fetches1.put(tp0, new FetchInfo(4, 1)); + fetches1.put(t3p0, new FetchInfo(0, 100)); + client.respondFrom(fetchResponse(fetches1), node); + + records = consumer.poll(Duration.ZERO); + + // should have t3 but not sent yet the t3 records + assertEquals(101, records.count()); + assertEquals(5L, consumer.position(tp0)); + assertEquals(100L, consumer.position(t3p0)); + + client.requests().clear(); + consumer.unsubscribe(); + consumer.close(); + } + @Test public void testGetGroupMetadata() { final Time time = new MockTime(); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java index 02971966f8478..7cde5b41e4674 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java @@ -38,6 +38,7 @@ import org.apache.kafka.common.errors.FencedInstanceIdException; import org.apache.kafka.common.errors.GroupAuthorizationException; import org.apache.kafka.common.errors.OffsetMetadataTooLarge; +import org.apache.kafka.common.errors.RebalanceInProgressException; import org.apache.kafka.common.errors.TopicAuthorizationException; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.internals.ClusterResourceListeners; @@ -1883,7 +1884,18 @@ public void testCommitOffsetRebalanceInProgress() { Map> memberSubscriptions = singletonMap(consumerId, singletonList(topic1)); partitionAssignor.prepare(singletonMap(consumerId, singletonList(t1p))); - client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE)); + coordinator.ensureActiveGroup(time.timer(0L)); + + assertTrue(coordinator.rejoinNeededOrPending()); + assertNull(coordinator.generationIfStable()); + + // when the state is REBALANCING, we would not even send out the request but fail immediately + assertThrows(RebalanceInProgressException.class, () -> coordinator.commitOffsetsSync(singletonMap(t1p, + new OffsetAndMetadata(100L, "metadata")), time.timer(Long.MAX_VALUE))); + + final Node coordinatorNode = new Node(Integer.MAX_VALUE - node.id(), node.host(), node.port()); + client.respondFrom(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE), coordinatorNode); + client.prepareResponse(body -> { SyncGroupRequest sync = (SyncGroupRequest) body; return sync.data.memberId().equals(consumerId) && @@ -1897,8 +1909,7 @@ public void testCommitOffsetRebalanceInProgress() { assertEquals(expectedGeneration, coordinator.generationIfStable()); prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.REBALANCE_IN_PROGRESS); - - assertThrows(CommitFailedException.class, () -> coordinator.commitOffsetsSync(singletonMap(t1p, + assertThrows(RebalanceInProgressException.class, () -> coordinator.commitOffsetsSync(singletonMap(t1p, new OffsetAndMetadata(100L, "metadata")), time.timer(Long.MAX_VALUE))); assertTrue(coordinator.rejoinNeededOrPending()); @@ -2336,7 +2347,7 @@ public void testConsumerRejoinAfterRebalance() { prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.REBALANCE_IN_PROGRESS); - assertThrows(CommitFailedException.class, () -> coordinator.commitOffsetsSync( + assertThrows(RebalanceInProgressException.class, () -> coordinator.commitOffsetsSync( singletonMap(t1p, new OffsetAndMetadata(100L)), time.timer(Long.MAX_VALUE))); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java index c3ce02a79a155..6e1e8a36b3690 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java @@ -199,7 +199,7 @@ public void topicSubscription() { assertEquals(1, state.subscription().size()); assertTrue(state.assignedPartitions().isEmpty()); assertEquals(0, state.numAssignedPartitions()); - assertTrue(state.partitionsAutoAssigned()); + assertTrue(state.hasAutoAssignedPartitions()); assertTrue(state.checkAssignmentMatchedSubscription(singleton(tp0))); state.assignFromSubscribed(singleton(tp0)); diff --git a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala index 9a40d69335b63..80b7156d1ab37 100644 --- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala @@ -20,6 +20,7 @@ package kafka.api import com.yammer.metrics.Metrics import com.yammer.metrics.core.Gauge import java.io.File +import java.util.Collections import java.util.concurrent.ExecutionException import kafka.admin.AclCommand @@ -215,7 +216,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas val consumer = createConsumer() consumer.assign(List(tp).asJava) consumeRecords(consumer, numRecords) - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } protected def confirmReauthenticationMetrics(): Unit = { @@ -244,7 +245,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas val consumer = createConsumer() consumer.subscribe(List(topic).asJava) consumeRecords(consumer, numRecords) - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } @Test @@ -255,7 +256,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas val consumer = createConsumer() consumer.subscribe(List(topic).asJava) consumeRecords(consumer, numRecords) - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } @Test @@ -266,7 +267,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas val consumer = createConsumer() consumer.subscribe(List(topic).asJava) consumeRecords(consumer, numRecords) - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } @Test @@ -277,7 +278,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas val consumer = createConsumer() consumer.assign(List(tp2).asJava) consumeRecords(consumer, numRecords, topic = tp2.topic) - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } private def setWildcardResourceAcls(): Unit = { @@ -351,12 +352,16 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas val e2 = intercept[ExecutionException] { adminClient.describeTopics(Set(topic).asJava).all().get() } assertTrue("Unexpected exception " + e2.getCause, e2.getCause.isInstanceOf[TopicAuthorizationException]) - // Verify that consumer manually assigning both authorized and unauthorized topic doesn't consume from either + // Verify that consumer manually assigning both authorized and unauthorized topic doesn't consume + // from the unauthorized topic and throw; since we can now return data during the time we are updating + // metadata / fetching positions, it is possible that the authorized topic record is returned during this time. consumer.assign(List(tp, tp2).asJava) sendRecords(producer, numRecords, tp2) + var topic2RecordConsumed = false def verifyNoRecords(records: ConsumerRecords[Array[Byte], Array[Byte]]): Boolean = { - assertTrue("Consumed records: " + records, records.isEmpty) - !records.isEmpty + assertEquals("Consumed records with unexpected partitions: " + records, Collections.singleton(tp2), records.partitions()) + topic2RecordConsumed = true + false } assertThrows[TopicAuthorizationException] { TestUtils.pollRecordsUntilTrue(consumer, verifyNoRecords, "Consumer didn't fail with authorization exception within timeout") @@ -364,9 +369,11 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas // Add ACLs and verify successful produce/consume/describe on first topic setReadAndWriteAcls(tp) - consumeRecordsIgnoreOneAuthorizationException(consumer, numRecords, startingOffset = 1, topic2) + if (!topic2RecordConsumed) { + consumeRecordsIgnoreOneAuthorizationException(consumer, numRecords, startingOffset = 1, topic2) + } sendRecords(producer, numRecords, tp) - consumeRecords(consumer, numRecords, topic = topic) + consumeRecordsIgnoreOneAuthorizationException(consumer, numRecords, startingOffset = 0, topic) val describeResults2 = adminClient.describeTopics(Set(topic, topic2).asJava).values assertEquals(1, describeResults2.get(topic).get().partitions().size()) assertEquals(1, describeResults2.get(topic2).get().partitions().size()) @@ -386,7 +393,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas case e: TopicAuthorizationException => assertEquals(Set(topic).asJava, e.unauthorizedTopics()) } - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } /** @@ -400,7 +407,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas consumer.assign(List(tp).asJava) // the exception is expected when the consumer attempts to lookup offsets consumeRecords(consumer) - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } @Test @@ -452,7 +459,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas case e: TopicAuthorizationException => assertEquals(Set(topic).asJava, e.unauthorizedTopics()) } - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } @Test @@ -468,7 +475,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas case e: TopicAuthorizationException => assertEquals(Set(topic).asJava, e.unauthorizedTopics()) } - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } private def noConsumeWithDescribeAclSetup(): Unit = { @@ -504,7 +511,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas case e: GroupAuthorizationException => assertEquals(group, e.groupId()) } - confirmReauthenticationMetrics + confirmReauthenticationMetrics() } protected final def sendRecords(producer: KafkaProducer[Array[Byte], Array[Byte]], diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index 4117355c145ee..a41a975c2e029 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -25,6 +25,7 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.AuthorizationException; import org.apache.kafka.common.errors.ProducerFencedException; +import org.apache.kafka.common.errors.RebalanceInProgressException; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.UnknownProducerIdException; import org.apache.kafka.common.errors.WakeupException; @@ -516,6 +517,11 @@ void commit(final boolean startNewTransaction, final Map p } } catch (final CommitFailedException | ProducerFencedException | UnknownProducerIdException error) { throw new TaskMigratedException(this, error); + } catch (final RebalanceInProgressException error) { + // commitSync throws this error and can be ignored (since EOS is not enabled, even if the task crashed + // immediately after this commit, we would just reprocess those records again) + log.info("Committing failed with a non-fatal error: {}, " + + "we can ignore this since commit may succeed still", error.toString()); } commitNeeded = false;