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 7f63e8bbc65d7..9c1a45ca5b125 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 @@ -888,7 +888,7 @@ private static Metrics buildMetrics(ConsumerConfig config, Time time, String cli public Set assignment() { acquireAndEnsureOpen(); try { - return Collections.unmodifiableSet(new HashSet<>(this.subscriptions.assignedPartitions())); + return Collections.unmodifiableSet(this.subscriptions.assignedPartitions()); } finally { release(); } @@ -1605,10 +1605,7 @@ public void seekToBeginning(Collection partitions) { acquireAndEnsureOpen(); try { Collection parts = partitions.size() == 0 ? this.subscriptions.assignedPartitions() : partitions; - for (TopicPartition tp : parts) { - log.info("Seeking to beginning of partition {}", tp); - subscriptions.requestOffsetReset(tp, OffsetResetStrategy.EARLIEST); - } + subscriptions.requestOffsetReset(parts, OffsetResetStrategy.EARLIEST); } finally { release(); } @@ -1633,10 +1630,7 @@ public void seekToEnd(Collection partitions) { acquireAndEnsureOpen(); try { Collection parts = partitions.size() == 0 ? this.subscriptions.assignedPartitions() : partitions; - for (TopicPartition tp : parts) { - log.info("Seeking to end of partition {}", tp); - subscriptions.requestOffsetReset(tp, OffsetResetStrategy.LATEST); - } + subscriptions.requestOffsetReset(parts, OffsetResetStrategy.LATEST); } finally { release(); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java index c8c2e72af100a..660a11238b3d2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java @@ -213,7 +213,7 @@ public synchronized ConsumerRecords poll(final Duration timeout) { public synchronized void addRecord(ConsumerRecord record) { ensureNotClosed(); TopicPartition tp = new TopicPartition(record.topic(), record.partition()); - Set currentAssigned = new HashSet<>(this.subscriptions.assignedPartitions()); + Set currentAssigned = this.subscriptions.assignedPartitions(); if (!currentAssigned.contains(tp)) throw new IllegalStateException("Cannot add records for a partition that is not assigned to the consumer"); List> recs = this.records.computeIfAbsent(tp, k -> new ArrayList<>()); @@ -312,8 +312,7 @@ public synchronized long position(TopicPartition partition, final Duration timeo @Override public synchronized void seekToBeginning(Collection partitions) { ensureNotClosed(); - for (TopicPartition tp : partitions) - subscriptions.requestOffsetReset(tp, OffsetResetStrategy.EARLIEST); + subscriptions.requestOffsetReset(partitions, OffsetResetStrategy.EARLIEST); } public synchronized void updateBeginningOffsets(Map newOffsets) { @@ -323,8 +322,7 @@ public synchronized void updateBeginningOffsets(Map newOff @Override public synchronized void seekToEnd(Collection partitions) { ensureNotClosed(); - for (TopicPartition tp : partitions) - subscriptions.requestOffsetReset(tp, OffsetResetStrategy.LATEST); + subscriptions.requestOffsetReset(partitions, OffsetResetStrategy.LATEST); } // needed for cases where you make a second call to endOffsets 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 6af36e93e59e4..f9655714cd17b 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 @@ -256,7 +256,7 @@ protected void onJoinComplete(int generation, return; } - Set assignedPartitions = new HashSet<>(subscriptions.assignedPartitions()); + Set assignedPartitions = subscriptions.assignedPartitions(); // The leader may have assigned partitions which match our subscription pattern, but which // were not explicitly requested, so we update the joined subscription here. @@ -463,8 +463,7 @@ protected void onJoinPrepare(int generation, String memberId) { // execute the user's callback before rebalance ConsumerRebalanceListener listener = subscriptions.rebalanceListener(); - // copy since about to be handed to user code - Set revoked = new HashSet<>(subscriptions.assignedPartitions()); + Set revoked = subscriptions.assignedPartitions(); log.info("Revoking previously assigned partitions {}", revoked); try { listener.onPartitionsRevoked(revoked); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index aa2936a3c1147..0460b3863a1eb 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -420,6 +420,15 @@ else if (strategy == OffsetResetStrategy.LATEST) return null; } + private OffsetResetStrategy timestampToOffsetResetStrategy(long timestamp) { + if (timestamp == ListOffsetRequest.EARLIEST_TIMESTAMP) + return OffsetResetStrategy.EARLIEST; + else if (timestamp == ListOffsetRequest.LATEST_TIMESTAMP) + return OffsetResetStrategy.LATEST; + else + return null; + } + /** * Reset offsets for all assigned partitions that require it. * @@ -664,22 +673,11 @@ private List> fetchRecords(PartitionRecords partitionRecord return emptyList(); } - private void resetOffsetIfNeeded(TopicPartition partition, Long requestedResetTimestamp, ListOffsetData offsetData) { - // we might lose the assignment while fetching the offset, or the user might seek to a different offset, - // so verify it is still assigned and still in need of the requested reset - if (!subscriptions.isAssigned(partition)) { - log.debug("Skipping reset of partition {} since it is no longer assigned", partition); - } else if (!subscriptions.isOffsetResetNeeded(partition)) { - log.debug("Skipping reset of partition {} since reset is no longer needed", partition); - } else if (!requestedResetTimestamp.equals(offsetResetStrategyTimestamp(partition))) { - log.debug("Skipping reset of partition {} since an alternative reset has been requested", partition); - } else { - SubscriptionState.FetchPosition position = new SubscriptionState.FetchPosition( - offsetData.offset, offsetData.leaderEpoch, metadata.leaderAndEpoch(partition)); - log.info("Resetting offset for partition {} to offset {}.", partition, position); - offsetData.leaderEpoch.ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(partition, epoch)); - subscriptions.seek(partition, position); - } + private void resetOffsetIfNeeded(TopicPartition partition, OffsetResetStrategy requestedResetStrategy, ListOffsetData offsetData) { + SubscriptionState.FetchPosition position = new SubscriptionState.FetchPosition( + offsetData.offset, offsetData.leaderEpoch, metadata.leaderAndEpoch(partition)); + offsetData.leaderEpoch.ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(partition, epoch)); + subscriptions.maybeSeek(partition, position.offset, requestedResetStrategy); } private void resetOffsetsAsync(Map partitionResetTimestamps) { @@ -703,7 +701,7 @@ public void onSuccess(ListOffsetResult result) { TopicPartition partition = fetchedOffset.getKey(); ListOffsetData offsetData = fetchedOffset.getValue(); ListOffsetRequest.PartitionData requestedReset = resetTimestamps.get(partition); - resetOffsetIfNeeded(partition, requestedReset.timestamp, offsetData); + resetOffsetIfNeeded(partition, timestampToOffsetResetStrategy(requestedReset.timestamp), offsetData); } } @@ -1728,7 +1726,7 @@ private void recordTopicFetchMetrics(String topic, int bytes, int records) { private void maybeUpdateAssignment(SubscriptionState subscription) { int newAssignmentId = subscription.assignmentId(); if (this.assignmentId != newAssignmentId) { - Set newAssignedPartitions = new HashSet<>(subscription.assignedPartitions()); + Set newAssignedPartitions = subscription.assignedPartitions(); for (TopicPartition tp : this.assignedPartitions) { if (!newAssignedPartitions.contains(tp)) { metrics.removeSensor(partitionLagMetricName(tp)); 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 4c87dba38204a..640ed5595ce97 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 @@ -38,7 +38,6 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.function.Predicate; import java.util.function.Supplier; import java.util.regex.Pattern; @@ -61,10 +60,7 @@ * Note that pause state as well as fetch/consumed positions are not preserved when partition * assignment is changed whether directly by the user or through a group rebalance. * - * Thread Safety: this class is generally not thread-safe. It should only be accessed in the - * consumer's calling thread. The only exception is {@link ConsumerMetadata} which accesses - * the subscription state needed to build and handle Metadata requests. The thread-safe methods - * are documented below. + * Thread Safety: this class is thread-safe. */ public class SubscriptionState { private static final String SUBSCRIPTION_EXCEPTION_MESSAGE = @@ -77,10 +73,10 @@ private enum SubscriptionType { } /* the type of subscription */ - private volatile SubscriptionType subscriptionType; + private SubscriptionType subscriptionType; /* the pattern user has requested */ - private volatile Pattern subscribedPattern; + private Pattern subscribedPattern; /* the list of topics the user has requested */ private Set subscription; @@ -88,7 +84,7 @@ private enum SubscriptionType { /* The list of topics the group has subscribed to. This may include some topics which are not part * of `subscription` for the leader of a group since it is responsible for detecting metadata changes * which require a group rebalance. */ - private final Set groupSubscription; + private Set groupSubscription; /* the partitions that are currently assigned, note that the order of partition matters (see FetchBuilder for more details) */ private final PartitionStates assignment; @@ -102,7 +98,7 @@ private enum SubscriptionType { private int assignmentId = 0; @Override - public String toString() { + public synchronized String toString() { return "SubscriptionState{" + "type=" + subscriptionType + ", subscribedPattern=" + subscribedPattern + @@ -112,7 +108,7 @@ public String toString() { ", assignment=" + assignment.partitionStateValues() + " (id=" + assignmentId + ")}"; } - public String prettyString() { + public synchronized String prettyString() { switch (subscriptionType) { case NONE: return "None"; @@ -130,9 +126,9 @@ public String prettyString() { public SubscriptionState(LogContext logContext, OffsetResetStrategy defaultResetStrategy) { this.log = logContext.logger(this.getClass()); this.defaultResetStrategy = defaultResetStrategy; - this.subscription = Collections.emptySet(); + this.subscription = new HashSet<>(); this.assignment = new PartitionStates<>(); - this.groupSubscription = ConcurrentHashMap.newKeySet(); + this.groupSubscription = new HashSet<>(); this.subscribedPattern = null; this.subscriptionType = SubscriptionType.NONE; } @@ -143,7 +139,7 @@ public SubscriptionState(LogContext logContext, OffsetResetStrategy defaultReset * * @return The current assignment Id */ - public int assignmentId() { + synchronized int assignmentId() { return assignmentId; } @@ -160,18 +156,19 @@ else if (this.subscriptionType != type) throw new IllegalStateException(SUBSCRIPTION_EXCEPTION_MESSAGE); } - public boolean subscribe(Set topics, ConsumerRebalanceListener listener) { - if (listener == null) - throw new IllegalArgumentException("RebalanceListener cannot be null"); - + public synchronized boolean subscribe(Set topics, ConsumerRebalanceListener listener) { + registerRebalanceListener(listener); setSubscriptionType(SubscriptionType.AUTO_TOPICS); - - this.rebalanceListener = listener; - return changeSubscription(topics); } - public boolean subscribeFromPattern(Set topics) { + public synchronized void subscribe(Pattern pattern, ConsumerRebalanceListener listener) { + registerRebalanceListener(listener); + setSubscriptionType(SubscriptionType.AUTO_PATTERN); + this.subscribedPattern = pattern; + } + + public synchronized boolean subscribeFromPattern(Set topics) { if (subscriptionType != SubscriptionType.AUTO_PATTERN) throw new IllegalArgumentException("Attempt to subscribe from pattern while subscription type set to " + subscriptionType); @@ -183,8 +180,9 @@ private boolean changeSubscription(Set topicsToSubscribe) { if (subscription.equals(topicsToSubscribe)) return false; - this.subscription = topicsToSubscribe; - this.groupSubscription.addAll(topicsToSubscribe); + subscription = topicsToSubscribe; + groupSubscription = new HashSet<>(groupSubscription); + groupSubscription.addAll(topicsToSubscribe); return true; } @@ -193,17 +191,18 @@ private boolean changeSubscription(Set topicsToSubscribe) { * that it receives metadata updates for all topics that the group is interested in. * @param topics The topics to add to the group subscription */ - public boolean groupSubscribe(Collection topics) { + synchronized boolean groupSubscribe(Collection topics) { if (!partitionsAutoAssigned()) throw new IllegalStateException(SUBSCRIPTION_EXCEPTION_MESSAGE); - return this.groupSubscription.addAll(topics); + groupSubscription = new HashSet<>(groupSubscription); + return groupSubscription.addAll(topics); } /** * Reset the group's subscription to only contain topics subscribed by this consumer. */ - public void resetGroupSubscription() { - this.groupSubscription.retainAll(subscription); + synchronized void resetGroupSubscription() { + groupSubscription = subscription; } /** @@ -211,7 +210,7 @@ public void resetGroupSubscription() { * note this is different from {@link #assignFromSubscribed(Collection)} * whose input partitions are provided from the subscribed topics. */ - public boolean assignFromUser(Set partitions) { + public synchronized boolean assignFromUser(Set partitions) { setSubscriptionType(SubscriptionType.USER_ASSIGNED); if (this.assignment.partitionSet().equals(partitions)) @@ -238,29 +237,28 @@ public boolean assignFromUser(Set partitions) { * * @return true if assignments matches subscription, otherwise false */ - public boolean assignFromSubscribed(Collection assignments) { + public synchronized boolean assignFromSubscribed(Collection assignments) { if (!this.partitionsAutoAssigned()) throw new IllegalArgumentException("Attempt to dynamically assign partitions while manual assignment in use"); - Predicate predicate = topicPartition -> { + boolean assignmentMatchedSubscription = true; + for (TopicPartition topicPartition : assignments) { if (this.subscribedPattern != null) { - boolean match = this.subscribedPattern.matcher(topicPartition.topic()).matches(); - if (!match) { + assignmentMatchedSubscription = this.subscribedPattern.matcher(topicPartition.topic()).matches(); + if (!assignmentMatchedSubscription) { log.info("Assigned partition {} for non-subscribed topic regex pattern; subscription pattern is {}", topicPartition, this.subscribedPattern); + break; } - return match; } else { - boolean match = this.subscription.contains(topicPartition.topic()); - if (!match) { + assignmentMatchedSubscription = this.subscription.contains(topicPartition.topic()); + if (!assignmentMatchedSubscription) { log.info("Assigned partition {} for non-subscribed topic; subscription is {}", topicPartition, this.subscription); + break; } - return match; } - }; - - boolean assignmentMatchedSubscription = assignments.stream().allMatch(predicate); + } if (assignmentMatchedSubscription) { Map assignedPartitionStates = partitionToStateMap( @@ -272,31 +270,27 @@ public boolean assignFromSubscribed(Collection assignments) { return assignmentMatchedSubscription; } - public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) { + private void registerRebalanceListener(ConsumerRebalanceListener listener) { if (listener == null) throw new IllegalArgumentException("RebalanceListener cannot be null"); - - setSubscriptionType(SubscriptionType.AUTO_PATTERN); - this.rebalanceListener = listener; - this.subscribedPattern = pattern; } /** - * Check whether pattern subscription is in use. This is thread-safe. + * Check whether pattern subscription is in use. * */ - public boolean hasPatternSubscription() { + synchronized boolean hasPatternSubscription() { return this.subscriptionType == SubscriptionType.AUTO_PATTERN; } - public boolean hasNoSubscriptionOrUserAssignment() { + public synchronized boolean hasNoSubscriptionOrUserAssignment() { return this.subscriptionType == SubscriptionType.NONE; } - public void unsubscribe() { + public synchronized void unsubscribe() { this.subscription = Collections.emptySet(); - this.groupSubscription.clear(); + this.groupSubscription = Collections.emptySet(); this.assignment.clear(); this.subscribedPattern = null; this.subscriptionType = SubscriptionType.NONE; @@ -306,18 +300,16 @@ public void unsubscribe() { /** * Check whether a topic matches a subscribed pattern. * - * This is thread-safe, but it may not always reflect the most recent subscription pattern. - * * @return true if pattern subscription is in use and the topic matches the subscribed pattern, false otherwise */ - public boolean matchesSubscribedPattern(String topic) { + synchronized boolean matchesSubscribedPattern(String topic) { Pattern pattern = this.subscribedPattern; if (hasPatternSubscription() && pattern != null) return pattern.matcher(topic).matches(); return false; } - public Set subscription() { + public synchronized Set subscription() { if (partitionsAutoAssigned()) return this.subscription; return Collections.emptySet(); @@ -335,19 +327,14 @@ public Set pausedPartitions() { * can do the partition assignment (which requires at least partition counts for all topics * to be assigned). * - * Note this is thread-safe since the Set is backed by a ConcurrentMap. - * * @return The union of all subscribed topics in the group if this member is the leader * of the current generation; otherwise it returns the same set as {@link #subscription()} */ - public Set groupSubscription() { + synchronized Set groupSubscription() { return this.groupSubscription; } - /** - * Note this is thread-safe since the Set is backed by a ConcurrentMap. - */ - public boolean isGroupSubscribed(String topic) { + synchronized boolean isGroupSubscribed(String topic) { return groupSubscription.contains(topic); } @@ -358,69 +345,87 @@ private TopicPartitionState assignedState(TopicPartition tp) { return state; } - public void seek(TopicPartition tp, FetchPosition position) { + private TopicPartitionState assignedStateOrNull(TopicPartition tp) { + return this.assignment.stateValue(tp); + } + + public synchronized void seek(TopicPartition tp, FetchPosition position) { assignedState(tp).seek(position); } - public void seekAndValidate(TopicPartition tp, FetchPosition position) { + public synchronized void seekAndValidate(TopicPartition tp, FetchPosition position) { assignedState(tp).seekAndValidate(position); } public void seek(TopicPartition tp, long offset) { - seek(tp, new FetchPosition(offset, Optional.empty(), new Metadata.LeaderAndEpoch(Node.noNode(), Optional.empty()))); + seek(tp, new FetchPosition(offset)); + } + + synchronized void maybeSeek(TopicPartition tp, long offset, OffsetResetStrategy requestedResetStrategy) { + TopicPartitionState state = assignedStateOrNull(tp); + if (state == null) { + log.debug("Skipping reset of partition {} since it is no longer assigned", tp); + } else if (!state.awaitingReset()) { + log.debug("Skipping reset of partition {} since reset is no longer needed", tp); + } else if (requestedResetStrategy != state.resetStrategy) { + log.debug("Skipping reset of partition {} since an alternative reset has been requested", tp); + } else { + log.info("Resetting offset for partition {} to offset {}.", tp, offset); + state.seek(new FetchPosition(offset)); + } } /** - * @return an unmodifiable view of the currently assigned partitions + * @return a modifiable copy of the currently assigned partitions */ - public Set assignedPartitions() { - return this.assignment.partitionSet(); + public synchronized Set assignedPartitions() { + return new HashSet<>(this.assignment.partitionSet()); } /** * Provides the number of assigned partitions in a thread safe manner. * @return the number of assigned partitions. */ - public int numAssignedPartitions() { + synchronized int numAssignedPartitions() { return this.assignment.size(); } - public List fetchablePartitions(Predicate isAvailable) { + synchronized List fetchablePartitions(Predicate isAvailable) { return assignment.stream() .filter(tpState -> isAvailable.test(tpState.topicPartition()) && tpState.value().isFetchable()) .map(PartitionStates.PartitionState::topicPartition) .collect(Collectors.toList()); } - public boolean partitionsAutoAssigned() { + synchronized boolean partitionsAutoAssigned() { return this.subscriptionType == SubscriptionType.AUTO_TOPICS || this.subscriptionType == SubscriptionType.AUTO_PATTERN; } - public void position(TopicPartition tp, FetchPosition position) { + public synchronized void position(TopicPartition tp, FetchPosition position) { assignedState(tp).position(position); } - public boolean maybeValidatePosition(TopicPartition tp, Metadata.LeaderAndEpoch leaderAndEpoch) { + synchronized boolean maybeValidatePosition(TopicPartition tp, Metadata.LeaderAndEpoch leaderAndEpoch) { return assignedState(tp).maybeValidatePosition(leaderAndEpoch); } - public boolean awaitingValidation(TopicPartition tp) { + synchronized boolean awaitingValidation(TopicPartition tp) { return assignedState(tp).awaitingValidation(); } - public void completeValidation(TopicPartition tp) { + public synchronized void completeValidation(TopicPartition tp) { assignedState(tp).validate(); } - public FetchPosition validPosition(TopicPartition tp) { + public synchronized FetchPosition validPosition(TopicPartition tp) { return assignedState(tp).validPosition(); } - public FetchPosition position(TopicPartition tp) { - return assignedState(tp).position(); + synchronized public FetchPosition position(TopicPartition tp) { + return assignedState(tp).position; } - public Long partitionLag(TopicPartition tp, IsolationLevel isolationLevel) { + synchronized Long partitionLag(TopicPartition tp, IsolationLevel isolationLevel) { TopicPartitionState topicPartitionState = assignedState(tp); if (isolationLevel == IsolationLevel.READ_COMMITTED) return topicPartitionState.lastStableOffset == null ? null : topicPartitionState.lastStableOffset - topicPartitionState.position.offset; @@ -428,21 +433,21 @@ public Long partitionLag(TopicPartition tp, IsolationLevel isolationLevel) { return topicPartitionState.highWatermark == null ? null : topicPartitionState.highWatermark - topicPartitionState.position.offset; } - public Long partitionLead(TopicPartition tp) { + synchronized Long partitionLead(TopicPartition tp) { TopicPartitionState topicPartitionState = assignedState(tp); return topicPartitionState.logStartOffset == null ? null : topicPartitionState.position.offset - topicPartitionState.logStartOffset; } - public void updateHighWatermark(TopicPartition tp, long highWatermark) { - assignedState(tp).highWatermark = highWatermark; + synchronized void updateHighWatermark(TopicPartition tp, long highWatermark) { + assignedState(tp).highWatermark(highWatermark); } - public void updateLogStartOffset(TopicPartition tp, long logStartOffset) { - assignedState(tp).logStartOffset = logStartOffset; + synchronized void updateLogStartOffset(TopicPartition tp, long logStartOffset) { + assignedState(tp).logStartOffset(logStartOffset); } - public void updateLastStableOffset(TopicPartition tp, long lastStableOffset) { - assignedState(tp).lastStableOffset = lastStableOffset; + synchronized void updateLastStableOffset(TopicPartition tp, long lastStableOffset) { + assignedState(tp).lastStableOffset(lastStableOffset); } /** @@ -453,7 +458,7 @@ public void updateLastStableOffset(TopicPartition tp, long lastStableOffset) { * @param preferredReadReplicaId The preferred read replica * @param timeMs The time at which this preferred replica is no longer valid */ - public void updatePreferredReadReplica(TopicPartition tp, int preferredReadReplicaId, Supplier timeMs) { + public synchronized void updatePreferredReadReplica(TopicPartition tp, int preferredReadReplicaId, Supplier timeMs) { assignedState(tp).updatePreferredReadReplica(preferredReadReplicaId, timeMs); } @@ -464,7 +469,7 @@ public void updatePreferredReadReplica(TopicPartition tp, int preferredReadRepli * @param timeMs The current time * @return Returns the current preferred read replica, if it has been set and if it has not expired. */ - public Optional preferredReadReplica(TopicPartition tp, long timeMs) { + public synchronized Optional preferredReadReplica(TopicPartition tp, long timeMs) { return assignedState(tp).preferredReadReplica(timeMs); } @@ -474,11 +479,11 @@ public Optional preferredReadReplica(TopicPartition tp, long timeMs) { * @param tp The topic partition * @return true if the preferred read replica was set, false otherwise. */ - public Optional clearPreferredReadReplica(TopicPartition tp) { + public synchronized Optional clearPreferredReadReplica(TopicPartition tp) { return assignedState(tp).clearPreferredReadReplica(); } - public Map allConsumed() { + public synchronized Map allConsumed() { Map allConsumed = new HashMap<>(); assignment.stream().forEach(state -> { TopicPartitionState partitionState = state.value(); @@ -489,41 +494,48 @@ public Map allConsumed() { return allConsumed; } - public void requestOffsetReset(TopicPartition partition, OffsetResetStrategy offsetResetStrategy) { + public synchronized void requestOffsetReset(TopicPartition partition, OffsetResetStrategy offsetResetStrategy) { assignedState(partition).reset(offsetResetStrategy); } + public synchronized void requestOffsetReset(Collection partitions, OffsetResetStrategy offsetResetStrategy) { + partitions.forEach(tp -> { + log.info("Seeking to {} offset of partition {}", offsetResetStrategy, tp); + assignedState(tp).reset(offsetResetStrategy); + }); + } + public void requestOffsetReset(TopicPartition partition) { requestOffsetReset(partition, defaultResetStrategy); } - public void setNextAllowedRetry(Set partitions, long nextAllowResetTimeMs) { + synchronized void setNextAllowedRetry(Set partitions, long nextAllowResetTimeMs) { for (TopicPartition partition : partitions) { assignedState(partition).setNextAllowedRetry(nextAllowResetTimeMs); } } - public boolean hasDefaultOffsetResetPolicy() { + boolean hasDefaultOffsetResetPolicy() { return defaultResetStrategy != OffsetResetStrategy.NONE; } - public boolean isOffsetResetNeeded(TopicPartition partition) { + public synchronized boolean isOffsetResetNeeded(TopicPartition partition) { return assignedState(partition).awaitingReset(); } - public OffsetResetStrategy resetStrategy(TopicPartition partition) { - return assignedState(partition).resetStrategy; + public synchronized OffsetResetStrategy resetStrategy(TopicPartition partition) { + return assignedState(partition).resetStrategy(); } - public boolean hasAllFetchPositions() { + public synchronized boolean hasAllFetchPositions() { return assignment.stream().allMatch(state -> state.value().hasValidPosition()); } - public Set missingFetchPositions() { + Set missingFetchPositions() { return collectPartitions(state -> !state.hasPosition(), Collectors.toSet()); } - private > T collectPartitions(Predicate filter, Collector collector) { + private synchronized > T collectPartitions(Predicate filter, Collector collector) { return assignment.stream() .filter(state -> filter.test(state.value())) .map(PartitionStates.PartitionState::topicPartition) @@ -531,7 +543,7 @@ private > T collectPartitions(Predicate partitionsWithNoOffsets = new HashSet<>(); assignment.stream().forEach(state -> { TopicPartition tp = state.topicPartition(); @@ -540,7 +552,7 @@ public void resetMissingPositions() { if (defaultResetStrategy == OffsetResetStrategy.NONE) partitionsWithNoOffsets.add(tp); else - partitionState.reset(defaultResetStrategy); + requestOffsetReset(tp); } }); @@ -548,50 +560,53 @@ public void resetMissingPositions() { throw new NoOffsetForPartitionException(partitionsWithNoOffsets); } - public Set partitionsNeedingReset(long nowMs) { + Set partitionsNeedingReset(long nowMs) { return collectPartitions(state -> state.awaitingReset() && !state.awaitingRetryBackoff(nowMs), Collectors.toSet()); } - public Set partitionsNeedingValidation(long nowMs) { + Set partitionsNeedingValidation(long nowMs) { return collectPartitions(state -> state.awaitingValidation() && !state.awaitingRetryBackoff(nowMs), Collectors.toSet()); } - public boolean isAssigned(TopicPartition tp) { + public synchronized boolean isAssigned(TopicPartition tp) { return assignment.contains(tp); } - public boolean isPaused(TopicPartition tp) { - return isAssigned(tp) && assignedState(tp).paused; + public synchronized boolean isPaused(TopicPartition tp) { + TopicPartitionState assignedOrNull = assignedStateOrNull(tp); + return assignedOrNull != null && assignedOrNull.isPaused(); } - public boolean isFetchable(TopicPartition tp) { - return isAssigned(tp) && assignedState(tp).isFetchable(); + synchronized boolean isFetchable(TopicPartition tp) { + TopicPartitionState assignedOrNull = assignedStateOrNull(tp); + return assignedOrNull != null && assignedOrNull.isFetchable(); } - public boolean hasValidPosition(TopicPartition tp) { - return isAssigned(tp) && assignedState(tp).hasValidPosition(); + public synchronized boolean hasValidPosition(TopicPartition tp) { + TopicPartitionState assignedOrNull = assignedStateOrNull(tp); + return assignedOrNull != null && assignedOrNull.hasValidPosition(); } - public void pause(TopicPartition tp) { + public synchronized void pause(TopicPartition tp) { assignedState(tp).pause(); } - public void resume(TopicPartition tp) { + public synchronized void resume(TopicPartition tp) { assignedState(tp).resume(); } - public void requestFailed(Set partitions, long nextRetryTimeMs) { + synchronized void requestFailed(Set partitions, long nextRetryTimeMs) { for (TopicPartition partition : partitions) assignedState(partition).requestFailed(nextRetryTimeMs); } - public void movePartitionToEnd(TopicPartition tp) { + synchronized void movePartitionToEnd(TopicPartition tp) { assignment.moveToEnd(tp); } - public ConsumerRebalanceListener rebalanceListener() { + public synchronized ConsumerRebalanceListener rebalanceListener() { return rebalanceListener; } @@ -773,10 +788,6 @@ private FetchPosition validPosition() { } } - private FetchPosition position() { - return position; - } - private void pause() { this.paused = true; } @@ -789,6 +800,21 @@ private boolean isFetchable() { return !paused && hasValidPosition(); } + private void highWatermark(Long highWatermark) { + this.highWatermark = highWatermark; + } + + private void logStartOffset(Long logStartOffset) { + this.logStartOffset = logStartOffset; + } + + private void lastStableOffset(Long lastStableOffset) { + this.lastStableOffset = lastStableOffset; + } + + private OffsetResetStrategy resetStrategy() { + return resetStrategy; + } } /** @@ -895,8 +921,12 @@ public boolean hasValidPosition() { */ public static class FetchPosition { public final long offset; - public final Optional offsetEpoch; - public final Metadata.LeaderAndEpoch currentLeader; + final Optional offsetEpoch; + final Metadata.LeaderAndEpoch currentLeader; + + FetchPosition(long offset) { + this(offset, Optional.empty(), new Metadata.LeaderAndEpoch(Node.noNode(), Optional.empty())); + } public FetchPosition(long offset, Optional offsetEpoch, Metadata.LeaderAndEpoch currentLeader) { this.offset = offset; @@ -908,7 +938,7 @@ public FetchPosition(long offset, Optional offsetEpoch, Metadata.Leader * Test if it is "safe" to fetch from a given leader and epoch. This effectively is testing if * {@link Metadata.LeaderAndEpoch} known to the subscription is equal to the one supplied by the caller. */ - public boolean safeToFetchFrom(Metadata.LeaderAndEpoch leaderAndEpoch) { + boolean safeToFetchFrom(Metadata.LeaderAndEpoch leaderAndEpoch) { return !currentLeader.leader.isEmpty() && currentLeader.equals(leaderAndEpoch); } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index 69036cafc5bb3..ee98e97d7e21b 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -109,6 +109,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -130,6 +131,8 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; @SuppressWarnings("deprecation") public class FetcherTest { @@ -1483,6 +1486,51 @@ public void testSeekWithInFlightReset() { assertEquals(237L, subscriptions.position(tp0).offset); } + @Test(timeout = 10000) + public void testEarlierOffsetResetArrivesLate() throws InterruptedException { + LogContext lc = new LogContext(); + buildFetcher(spy(new SubscriptionState(lc, OffsetResetStrategy.EARLIEST)), lc); + assignFromUser(singleton(tp0)); + + ExecutorService es = Executors.newSingleThreadExecutor(); + CountDownLatch latchLatestStart = new CountDownLatch(1); + CountDownLatch latchEarliestStart = new CountDownLatch(1); + CountDownLatch latchEarliestDone = new CountDownLatch(1); + CountDownLatch latchEarliestFinish = new CountDownLatch(1); + try { + doAnswer(invocation -> { + latchLatestStart.countDown(); + latchEarliestStart.await(); + Object result = invocation.callRealMethod(); + latchEarliestDone.countDown(); + return result; + }).when(subscriptions).maybeSeek(tp0, 0L, OffsetResetStrategy.EARLIEST); + + es.submit(() -> { + subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.EARLIEST); + fetcher.resetOffsetsIfNeeded(); + consumerClient.pollNoWakeup(); + client.respond(listOffsetResponse(Errors.NONE, 1L, 0L)); + consumerClient.pollNoWakeup(); + latchEarliestFinish.countDown(); + }, Void.class); + + latchLatestStart.await(); + subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST); + fetcher.resetOffsetsIfNeeded(); + consumerClient.pollNoWakeup(); + client.respond(listOffsetResponse(Errors.NONE, 1L, 10L)); + latchEarliestStart.countDown(); + latchEarliestDone.await(); + consumerClient.pollNoWakeup(); + latchEarliestFinish.await(); + assertEquals(10, subscriptions.position(tp0).offset); + } finally { + es.shutdown(); + es.awaitTermination(10000, TimeUnit.MILLISECONDS); + } + } + @Test public void testChangeResetWithInFlightReset() { buildFetcher(); @@ -2816,7 +2864,8 @@ public void testFetcherConcurrency() throws Exception { for (int i = 0; i < numPartitions; i++) topicPartitions.add(new TopicPartition(topicName, i)); - buildDependencies(new MetricConfig(), OffsetResetStrategy.EARLIEST, Long.MAX_VALUE); + LogContext logContext = new LogContext(); + buildDependencies(new MetricConfig(), Long.MAX_VALUE, new SubscriptionState(logContext, OffsetResetStrategy.EARLIEST), logContext); fetcher = new Fetcher( new LogContext(), @@ -3574,7 +3623,26 @@ private void buildFetcher(MetricConfig metricConfig, int maxPollRecords, IsolationLevel isolationLevel, long metadataExpireMs) { - buildDependencies(metricConfig, offsetResetStrategy, metadataExpireMs); + LogContext logContext = new LogContext(); + SubscriptionState subscriptionState = new SubscriptionState(logContext, offsetResetStrategy); + buildFetcher(metricConfig, keyDeserializer, valueDeserializer, maxPollRecords, isolationLevel, metadataExpireMs, + subscriptionState, logContext); + } + + private void buildFetcher(SubscriptionState subscriptionState, LogContext logContext) { + buildFetcher(new MetricConfig(), new ByteArrayDeserializer(), new ByteArrayDeserializer(), Integer.MAX_VALUE, + IsolationLevel.READ_UNCOMMITTED, Long.MAX_VALUE, subscriptionState, logContext); + } + + private void buildFetcher(MetricConfig metricConfig, + Deserializer keyDeserializer, + Deserializer valueDeserializer, + int maxPollRecords, + IsolationLevel isolationLevel, + long metadataExpireMs, + SubscriptionState subscriptionState, + LogContext logContext) { + buildDependencies(metricConfig, metadataExpireMs, subscriptionState, logContext); fetcher = new Fetcher<>( new LogContext(), consumerClient, @@ -3598,11 +3666,12 @@ private void buildFetcher(MetricConfig metricConfig, apiVersions); } - - private void buildDependencies(MetricConfig metricConfig, OffsetResetStrategy offsetResetStrategy, long metadataExpireMs) { - LogContext logContext = new LogContext(); + private void buildDependencies(MetricConfig metricConfig, + long metadataExpireMs, + SubscriptionState subscriptionState, + LogContext logContext) { time = new MockTime(1); - subscriptions = new SubscriptionState(logContext, offsetResetStrategy); + subscriptions = subscriptionState; metadata = new ConsumerMetadata(0, metadataExpireMs, false, false, subscriptions, logContext, new ClusterResourceListeners()); client = new MockClient(time, metadata);