From 9acfc8ca438032baffe662ee8d6096c02c4f4016 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Thu, 7 Mar 2019 14:54:39 +0100 Subject: [PATCH 1/7] KAFKA-7703: position() may return a wrong offset after "seekToEnd" is called --- .../kafka/clients/consumer/KafkaConsumer.java | 12 +- .../kafka/clients/consumer/MockConsumer.java | 8 +- .../internals/ConsumerCoordinator.java | 5 +- .../clients/consumer/internals/Fetcher.java | 21 +- .../consumer/internals/SubscriptionState.java | 230 +++++++++++------- .../common/internals/PartitionStates.java | 6 +- .../consumer/internals/FetcherTest.java | 49 +++- gradle/spotbugs-exclude.xml | 8 + 8 files changed, 219 insertions(+), 120 deletions(-) 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..b2eff75f69956 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,21 +673,17 @@ private List> fetchRecords(PartitionRecords partitionRecord return emptyList(); } - private void resetOffsetIfNeeded(TopicPartition partition, Long requestedResetTimestamp, ListOffsetData offsetData) { + private void resetOffsetIfNeeded(TopicPartition partition, OffsetResetStrategy requestedResetStrategy, 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); + subscriptions.maybeSeek(partition, position.offset, requestedResetStrategy); } } @@ -703,7 +708,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 +1733,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..30f55cb5ea89c 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,8 @@ * 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 generally thread-safe. The only exception is the consumer rebalance + * listener which is guaranteed to be executed without holding any locks. */ public class SubscriptionState { private static final String SUBSCRIPTION_EXCEPTION_MESSAGE = @@ -77,10 +74,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 +85,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; @@ -130,9 +127,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 +140,7 @@ public SubscriptionState(LogContext logContext, OffsetResetStrategy defaultReset * * @return The current assignment Id */ - public int assignmentId() { + synchronized int assignmentId() { return assignmentId; } @@ -161,17 +158,15 @@ else if (this.subscriptionType != type) } public boolean subscribe(Set topics, ConsumerRebalanceListener listener) { - if (listener == null) - throw new IllegalArgumentException("RebalanceListener cannot be null"); - - setSubscriptionType(SubscriptionType.AUTO_TOPICS); - - this.rebalanceListener = listener; + registerRebalanceListener(listener); - return changeSubscription(topics); + synchronized (this) { + setSubscriptionType(SubscriptionType.AUTO_TOPICS); + return changeSubscription(topics); + } } - public boolean subscribeFromPattern(Set topics) { + 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 +178,8 @@ private boolean changeSubscription(Set topicsToSubscribe) { if (subscription.equals(topicsToSubscribe)) return false; - this.subscription = topicsToSubscribe; - this.groupSubscription.addAll(topicsToSubscribe); + subscription = topicsToSubscribe; + groupSubscription = topicsToSubscribe; return true; } @@ -193,17 +188,19 @@ 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 = new HashSet<>(groupSubscription); + groupSubscription.retainAll(subscription); } /** @@ -211,7 +208,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,7 +235,7 @@ 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"); @@ -273,30 +270,35 @@ public boolean assignFromSubscribed(Collection assignments) { } public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) { - if (listener == null) - throw new IllegalArgumentException("RebalanceListener cannot be null"); + registerRebalanceListener(listener); - setSubscriptionType(SubscriptionType.AUTO_PATTERN); + synchronized (this) { + setSubscriptionType(SubscriptionType.AUTO_PATTERN); + this.subscribedPattern = pattern; + } + } + private void registerRebalanceListener(ConsumerRebalanceListener listener) { + if (listener == null) + throw new IllegalArgumentException("RebalanceListener cannot be null"); 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 +308,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,29 +335,32 @@ 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); } - private TopicPartitionState assignedState(TopicPartition tp) { + private synchronized TopicPartitionState assignedState(TopicPartition tp) { + return assignedStateWithoutLock(tp); + } + + private TopicPartitionState assignedStateWithoutLock(TopicPartition tp) { TopicPartitionState state = this.assignment.stateValue(tp); if (state == null) throw new IllegalStateException("No current assignment for partition " + tp); return state; } + private synchronized TopicPartitionState assignedStateOrNull(TopicPartition tp) { + return this.assignment.stateValue(tp); + } + public void seek(TopicPartition tp, FetchPosition position) { assignedState(tp).seek(position); } @@ -370,29 +373,37 @@ public void seek(TopicPartition tp, long offset) { seek(tp, new FetchPosition(offset, Optional.empty(), new Metadata.LeaderAndEpoch(Node.noNode(), Optional.empty()))); } + void maybeSeek(TopicPartition tp, long offset, OffsetResetStrategy requestedResetStrategy) { + if (!assignedState(tp).maybeSeek(new FetchPosition(offset, Optional.empty(), new Metadata.LeaderAndEpoch(Node.noNode(), Optional.empty())), requestedResetStrategy)) + log.debug("Skipping reset of partition {} to offset {} " + + "since it is no longer needed or an alternative reset has been requested", tp, offset); + else + log.info("Resetting offset for partition {} to offset {}.", tp, 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) { + 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; } @@ -417,10 +428,10 @@ public FetchPosition validPosition(TopicPartition tp) { } public FetchPosition position(TopicPartition tp) { - return assignedState(tp).position(); + return assignedState(tp).position; } - public Long partitionLag(TopicPartition tp, IsolationLevel isolationLevel) { + 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; @@ -433,16 +444,16 @@ public Long partitionLead(TopicPartition tp) { return topicPartitionState.logStartOffset == null ? null : topicPartitionState.position.offset - topicPartitionState.logStartOffset; } - public void updateHighWatermark(TopicPartition tp, long highWatermark) { - assignedState(tp).highWatermark = highWatermark; + void updateHighWatermark(TopicPartition tp, long highWatermark) { + assignedState(tp).highWatermark(highWatermark); } - public void updateLogStartOffset(TopicPartition tp, long logStartOffset) { - assignedState(tp).logStartOffset = logStartOffset; + void updateLogStartOffset(TopicPartition tp, long logStartOffset) { + assignedState(tp).logStartOffset(logStartOffset); } - public void updateLastStableOffset(TopicPartition tp, long lastStableOffset) { - assignedState(tp).lastStableOffset = lastStableOffset; + void updateLastStableOffset(TopicPartition tp, long lastStableOffset) { + assignedState(tp).lastStableOffset(lastStableOffset); } /** @@ -478,7 +489,7 @@ public 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(); @@ -493,17 +504,24 @@ public void requestOffsetReset(TopicPartition partition, OffsetResetStrategy off assignedState(partition).reset(offsetResetStrategy); } + public synchronized void requestOffsetReset(Collection partitions, OffsetResetStrategy offsetResetStrategy) { + partitions.forEach(tp -> { + log.info("Seeking to {} offset of partition {}", offsetResetStrategy, tp); + assignedStateWithoutLock(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); + assignedStateWithoutLock(partition).setNextAllowedRetry(nextAllowResetTimeMs); } } - public boolean hasDefaultOffsetResetPolicy() { + boolean hasDefaultOffsetResetPolicy() { return defaultResetStrategy != OffsetResetStrategy.NONE; } @@ -512,18 +530,18 @@ public boolean isOffsetResetNeeded(TopicPartition partition) { } public OffsetResetStrategy resetStrategy(TopicPartition partition) { - return assignedState(partition).resetStrategy; + 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 +549,7 @@ private > T collectPartitions(Predicate partitionsWithNoOffsets = new HashSet<>(); assignment.stream().forEach(state -> { TopicPartition tp = state.topicPartition(); @@ -540,7 +558,7 @@ public void resetMissingPositions() { if (defaultResetStrategy == OffsetResetStrategy.NONE) partitionsWithNoOffsets.add(tp); else - partitionState.reset(defaultResetStrategy); + requestOffsetReset(tp); } }); @@ -548,7 +566,7 @@ 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()); } @@ -558,20 +576,23 @@ public Set partitionsNeedingValidation(long 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; + TopicPartitionState assignedOrNull = assignedStateOrNull(tp); + return assignedOrNull != null && assignedOrNull.isPaused(); } - public boolean isFetchable(TopicPartition tp) { - return isAssigned(tp) && assignedState(tp).isFetchable(); + boolean isFetchable(TopicPartition tp) { + TopicPartitionState assignedOrNull = assignedStateOrNull(tp); + return assignedOrNull != null && assignedOrNull.isFetchable(); } public boolean hasValidPosition(TopicPartition tp) { - return isAssigned(tp) && assignedState(tp).hasValidPosition(); + TopicPartitionState assignedOrNull = assignedStateOrNull(tp); + return assignedOrNull != null && assignedOrNull.hasValidPosition(); } public void pause(TopicPartition tp) { @@ -582,12 +603,12 @@ public 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); + assignedStateWithoutLock(partition).requestFailed(nextRetryTimeMs); } - public void movePartitionToEnd(TopicPartition tp) { + synchronized void movePartitionToEnd(TopicPartition tp) { assignment.moveToEnd(tp); } @@ -670,6 +691,10 @@ private void reset(OffsetResetStrategy strategy) { }); } + private synchronized boolean isResetAllowed(long nowMs) { + return nextRetryTimeMs == null || nowMs >= nextRetryTimeMs; + } + private boolean maybeValidatePosition(Metadata.LeaderAndEpoch currentLeaderAndEpoch) { if (this.fetchState.equals(FetchStates.AWAIT_RESET)) { return false; @@ -722,27 +747,27 @@ private boolean awaitingRetryBackoff(long nowMs) { return nextRetryTimeMs != null && nowMs < nextRetryTimeMs; } - private boolean awaitingReset() { + private synchronized boolean awaitingReset() { return fetchState.equals(FetchStates.AWAIT_RESET); } - private void setNextAllowedRetry(long nextAllowedRetryTimeMs) { + private synchronized void setNextAllowedRetry(long nextAllowedRetryTimeMs) { this.nextRetryTimeMs = nextAllowedRetryTimeMs; } - private void requestFailed(long nextAllowedRetryTimeMs) { + private synchronized void requestFailed(long nextAllowedRetryTimeMs) { this.nextRetryTimeMs = nextAllowedRetryTimeMs; } - private boolean hasValidPosition() { + private synchronized boolean hasValidPosition() { return fetchState.hasValidPosition(); } - private boolean hasPosition() { + private synchronized boolean hasPosition() { return fetchState.hasPosition(); } - private boolean isPaused() { + private synchronized boolean isPaused() { return paused; } @@ -759,7 +784,15 @@ private void seekAndValidate(FetchPosition fetchPosition) { validatePosition(fetchPosition); } - private void position(FetchPosition position) { + private synchronized boolean maybeSeek(FetchPosition fetchPosition, OffsetResetStrategy requestedResetStrategy) { + if (this.resetStrategy == null || this.resetStrategy != requestedResetStrategy) + return false; + + seek(fetchPosition); + return true; + } + + private synchronized void position(FetchPosition position) { if (!hasValidPosition()) throw new IllegalStateException("Cannot set a new position without a valid current position"); this.position = position; @@ -777,18 +810,33 @@ private FetchPosition position() { return position; } - private void pause() { + private synchronized void pause() { this.paused = true; } - private void resume() { + private synchronized void resume() { this.paused = false; } - private boolean isFetchable() { + private synchronized boolean isFetchable() { return !paused && hasValidPosition(); } + private synchronized void highWatermark(Long highWatermark) { + this.highWatermark = highWatermark; + } + + private synchronized void logStartOffset(Long logStartOffset) { + this.logStartOffset = logStartOffset; + } + + private synchronized void lastStableOffset(Long lastStableOffset) { + this.lastStableOffset = lastStableOffset; + } + + private synchronized OffsetResetStrategy resetStrategy() { + return resetStrategy; + } } /** diff --git a/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java b/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java index 22e183ab23865..982ba5aa5ccfb 100644 --- a/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java +++ b/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java @@ -43,11 +43,11 @@ */ public class PartitionStates { - private final LinkedHashMap map = new LinkedHashMap<>(); + private final Map map = new LinkedHashMap<>(); private final Set partitionSetView = Collections.unmodifiableSet(map.keySet()); /* the number of partitions that are currently assigned available in a thread safe manner */ - private volatile int size = 0; + private int size = 0; public PartitionStates() {} @@ -107,7 +107,7 @@ public LinkedHashMap partitionStateMap() { /** * Returns the partition state values in order. */ - public List partitionStateValues() { + List partitionStateValues() { return new ArrayList<>(map.values()); } 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..118b9a5e3bcd5 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,50 @@ public void testSeekWithInFlightReset() { assertEquals(237L, subscriptions.position(tp0).offset); } + @Test(timeout = 10000) + public void testEarlierOffsetResetArrivesLate() throws InterruptedException { + buildFetcher(); + 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(new Long(10), subscriptions.position(tp0)); + } finally { + es.shutdown(); + es.awaitTermination(10000, TimeUnit.MILLISECONDS); + } + } + @Test public void testChangeResetWithInFlightReset() { buildFetcher(); @@ -3602,7 +3649,7 @@ private void buildFetcher(MetricConfig metricConfig, private void buildDependencies(MetricConfig metricConfig, OffsetResetStrategy offsetResetStrategy, long metadataExpireMs) { LogContext logContext = new LogContext(); time = new MockTime(1); - subscriptions = new SubscriptionState(logContext, offsetResetStrategy); + subscriptions = spy(new SubscriptionState(logContext, offsetResetStrategy)); metadata = new ConsumerMetadata(0, metadataExpireMs, false, false, subscriptions, logContext, new ClusterResourceListeners()); client = new MockClient(time, metadata); diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml index 721b05e464c0e..c97eeeb68a817 100644 --- a/gradle/spotbugs-exclude.xml +++ b/gradle/spotbugs-exclude.xml @@ -272,6 +272,14 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read + + + + + + From d2d4cca1a3fbc9d544d4bb504fd4e3c45614cea8 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Thu, 25 Apr 2019 15:29:52 +0200 Subject: [PATCH 2/7] Synchronize SubscriptionState instead of TopicPartitionState --- .../clients/consumer/internals/Fetcher.java | 16 +- .../consumer/internals/SubscriptionState.java | 160 ++++++++---------- .../common/internals/PartitionStates.java | 2 +- .../consumer/internals/FetcherTest.java | 2 +- 4 files changed, 79 insertions(+), 101 deletions(-) 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 b2eff75f69956..96747174b42ae 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 @@ -674,17 +674,11 @@ private List> fetchRecords(PartitionRecords partitionRecord } private void resetOffsetIfNeeded(TopicPartition partition, OffsetResetStrategy requestedResetStrategy, 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 { - 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.maybeSeek(partition, position.offset, requestedResetStrategy); - } + 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.maybeSeek(partition, position.offset, requestedResetStrategy); } private void resetOffsetsAsync(Map partitionResetTimestamps) { 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 30f55cb5ea89c..d6bf473d4705e 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 @@ -157,13 +157,16 @@ else if (this.subscriptionType != type) throw new IllegalStateException(SUBSCRIPTION_EXCEPTION_MESSAGE); } - public boolean subscribe(Set topics, ConsumerRebalanceListener listener) { + public synchronized boolean subscribe(Set topics, ConsumerRebalanceListener listener) { registerRebalanceListener(listener); + setSubscriptionType(SubscriptionType.AUTO_TOPICS); + return changeSubscription(topics); + } - synchronized (this) { - setSubscriptionType(SubscriptionType.AUTO_TOPICS); - return changeSubscription(topics); - } + public synchronized void subscribe(Pattern pattern, ConsumerRebalanceListener listener) { + registerRebalanceListener(listener); + setSubscriptionType(SubscriptionType.AUTO_PATTERN); + this.subscribedPattern = pattern; } public synchronized boolean subscribeFromPattern(Set topics) { @@ -269,15 +272,6 @@ public synchronized boolean assignFromSubscribed(Collection assi return assignmentMatchedSubscription; } - public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) { - registerRebalanceListener(listener); - - synchronized (this) { - setSubscriptionType(SubscriptionType.AUTO_PATTERN); - this.subscribedPattern = pattern; - } - } - private void registerRebalanceListener(ConsumerRebalanceListener listener) { if (listener == null) throw new IllegalArgumentException("RebalanceListener cannot be null"); @@ -346,39 +340,41 @@ synchronized boolean isGroupSubscribed(String topic) { return groupSubscription.contains(topic); } - private synchronized TopicPartitionState assignedState(TopicPartition tp) { - return assignedStateWithoutLock(tp); - } - - private TopicPartitionState assignedStateWithoutLock(TopicPartition tp) { + private TopicPartitionState assignedState(TopicPartition tp) { TopicPartitionState state = this.assignment.stateValue(tp); if (state == null) throw new IllegalStateException("No current assignment for partition " + tp); return state; } - private synchronized TopicPartitionState assignedStateOrNull(TopicPartition tp) { + private TopicPartitionState assignedStateOrNull(TopicPartition tp) { return this.assignment.stateValue(tp); } - public void seek(TopicPartition tp, FetchPosition position) { + 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()))); - } - - void maybeSeek(TopicPartition tp, long offset, OffsetResetStrategy requestedResetStrategy) { - if (!assignedState(tp).maybeSeek(new FetchPosition(offset, Optional.empty(), new Metadata.LeaderAndEpoch(Node.noNode(), Optional.empty())), requestedResetStrategy)) - log.debug("Skipping reset of partition {} to offset {} " + - "since it is no longer needed or an alternative reset has been requested", tp, offset); - else + 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)); + } } /** @@ -396,7 +392,7 @@ synchronized int numAssignedPartitions() { return this.assignment.size(); } - 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) @@ -407,31 +403,31 @@ 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) { + synchronized public FetchPosition position(TopicPartition tp) { return assignedState(tp).position; } - 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; @@ -439,20 +435,20 @@ 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; } - void updateHighWatermark(TopicPartition tp, long highWatermark) { + synchronized void updateHighWatermark(TopicPartition tp, long highWatermark) { assignedState(tp).highWatermark(highWatermark); } - void updateLogStartOffset(TopicPartition tp, long logStartOffset) { + synchronized void updateLogStartOffset(TopicPartition tp, long logStartOffset) { assignedState(tp).logStartOffset(logStartOffset); } - void updateLastStableOffset(TopicPartition tp, long lastStableOffset) { + synchronized void updateLastStableOffset(TopicPartition tp, long lastStableOffset) { assignedState(tp).lastStableOffset(lastStableOffset); } @@ -500,14 +496,14 @@ public synchronized 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); - assignedStateWithoutLock(tp).reset(offsetResetStrategy); + assignedState(tp).reset(offsetResetStrategy); }); } @@ -517,7 +513,7 @@ public void requestOffsetReset(TopicPartition partition) { synchronized void setNextAllowedRetry(Set partitions, long nextAllowResetTimeMs) { for (TopicPartition partition : partitions) { - assignedStateWithoutLock(partition).setNextAllowedRetry(nextAllowResetTimeMs); + assignedState(partition).setNextAllowedRetry(nextAllowResetTimeMs); } } @@ -525,11 +521,11 @@ 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) { + public synchronized OffsetResetStrategy resetStrategy(TopicPartition partition) { return assignedState(partition).resetStrategy(); } @@ -571,7 +567,7 @@ Set partitionsNeedingReset(long nowMs) { Collectors.toSet()); } - public Set partitionsNeedingValidation(long nowMs) { + Set partitionsNeedingValidation(long nowMs) { return collectPartitions(state -> state.awaitingValidation() && !state.awaitingRetryBackoff(nowMs), Collectors.toSet()); } @@ -580,39 +576,39 @@ public synchronized boolean isAssigned(TopicPartition tp) { return assignment.contains(tp); } - public boolean isPaused(TopicPartition tp) { + public synchronized boolean isPaused(TopicPartition tp) { TopicPartitionState assignedOrNull = assignedStateOrNull(tp); return assignedOrNull != null && assignedOrNull.isPaused(); } - boolean isFetchable(TopicPartition tp) { + synchronized boolean isFetchable(TopicPartition tp) { TopicPartitionState assignedOrNull = assignedStateOrNull(tp); return assignedOrNull != null && assignedOrNull.isFetchable(); } - public boolean hasValidPosition(TopicPartition tp) { + 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(); } synchronized void requestFailed(Set partitions, long nextRetryTimeMs) { for (TopicPartition partition : partitions) - assignedStateWithoutLock(partition).requestFailed(nextRetryTimeMs); + assignedState(partition).requestFailed(nextRetryTimeMs); } synchronized void movePartitionToEnd(TopicPartition tp) { assignment.moveToEnd(tp); } - public ConsumerRebalanceListener rebalanceListener() { + public synchronized ConsumerRebalanceListener rebalanceListener() { return rebalanceListener; } @@ -691,10 +687,6 @@ private void reset(OffsetResetStrategy strategy) { }); } - private synchronized boolean isResetAllowed(long nowMs) { - return nextRetryTimeMs == null || nowMs >= nextRetryTimeMs; - } - private boolean maybeValidatePosition(Metadata.LeaderAndEpoch currentLeaderAndEpoch) { if (this.fetchState.equals(FetchStates.AWAIT_RESET)) { return false; @@ -747,27 +739,27 @@ private boolean awaitingRetryBackoff(long nowMs) { return nextRetryTimeMs != null && nowMs < nextRetryTimeMs; } - private synchronized boolean awaitingReset() { + private boolean awaitingReset() { return fetchState.equals(FetchStates.AWAIT_RESET); } - private synchronized void setNextAllowedRetry(long nextAllowedRetryTimeMs) { + private void setNextAllowedRetry(long nextAllowedRetryTimeMs) { this.nextRetryTimeMs = nextAllowedRetryTimeMs; } - private synchronized void requestFailed(long nextAllowedRetryTimeMs) { + private void requestFailed(long nextAllowedRetryTimeMs) { this.nextRetryTimeMs = nextAllowedRetryTimeMs; } - private synchronized boolean hasValidPosition() { + private boolean hasValidPosition() { return fetchState.hasValidPosition(); } - private synchronized boolean hasPosition() { + private boolean hasPosition() { return fetchState.hasPosition(); } - private synchronized boolean isPaused() { + private boolean isPaused() { return paused; } @@ -784,15 +776,7 @@ private void seekAndValidate(FetchPosition fetchPosition) { validatePosition(fetchPosition); } - private synchronized boolean maybeSeek(FetchPosition fetchPosition, OffsetResetStrategy requestedResetStrategy) { - if (this.resetStrategy == null || this.resetStrategy != requestedResetStrategy) - return false; - - seek(fetchPosition); - return true; - } - - private synchronized void position(FetchPosition position) { + private void position(FetchPosition position) { if (!hasValidPosition()) throw new IllegalStateException("Cannot set a new position without a valid current position"); this.position = position; @@ -806,35 +790,31 @@ private FetchPosition validPosition() { } } - private FetchPosition position() { - return position; - } - - private synchronized void pause() { + private void pause() { this.paused = true; } - private synchronized void resume() { + private void resume() { this.paused = false; } - private synchronized boolean isFetchable() { + private boolean isFetchable() { return !paused && hasValidPosition(); } - private synchronized void highWatermark(Long highWatermark) { + private void highWatermark(Long highWatermark) { this.highWatermark = highWatermark; } - private synchronized void logStartOffset(Long logStartOffset) { + private void logStartOffset(Long logStartOffset) { this.logStartOffset = logStartOffset; } - private synchronized void lastStableOffset(Long lastStableOffset) { + private void lastStableOffset(Long lastStableOffset) { this.lastStableOffset = lastStableOffset; } - private synchronized OffsetResetStrategy resetStrategy() { + private OffsetResetStrategy resetStrategy() { return resetStrategy; } } @@ -943,8 +923,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; @@ -956,7 +940,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/main/java/org/apache/kafka/common/internals/PartitionStates.java b/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java index 982ba5aa5ccfb..8913a4343c87e 100644 --- a/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java +++ b/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java @@ -107,7 +107,7 @@ public LinkedHashMap partitionStateMap() { /** * Returns the partition state values in order. */ - List partitionStateValues() { + public List partitionStateValues() { return new ArrayList<>(map.values()); } 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 118b9a5e3bcd5..9028e7fae08f8 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 @@ -1523,7 +1523,7 @@ public void testEarlierOffsetResetArrivesLate() throws InterruptedException { latchEarliestDone.await(); consumerClient.pollNoWakeup(); latchEarliestFinish.await(); - assertEquals(new Long(10), subscriptions.position(tp0)); + assertEquals(10, subscriptions.position(tp0).offset); } finally { es.shutdown(); es.awaitTermination(10000, TimeUnit.MILLISECONDS); From f01800d01db9df6d02220c2f38bdfa6595c18a76 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Tue, 21 May 2019 10:44:36 +0200 Subject: [PATCH 3/7] Synchronize preferred replica methods --- .../kafka/clients/consumer/internals/SubscriptionState.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 d6bf473d4705e..cfd4929bbb26d 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 @@ -460,7 +460,7 @@ synchronized void updateLastStableOffset(TopicPartition tp, long lastStableOffse * @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); } @@ -471,7 +471,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); } @@ -481,7 +481,7 @@ 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(); } From 818389b3a4ebea87458daad8edb7fb36caabf9f0 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Thu, 23 May 2019 11:35:52 +0200 Subject: [PATCH 4/7] Fix review comments --- .../clients/consumer/internals/Fetcher.java | 1 - .../consumer/internals/SubscriptionState.java | 22 +++++++++---------- .../common/internals/PartitionStates.java | 4 ++-- gradle/spotbugs-exclude.xml | 8 ------- 4 files changed, 12 insertions(+), 23 deletions(-) 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 96747174b42ae..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 @@ -676,7 +676,6 @@ private List> fetchRecords(PartitionRecords partitionRecord private void resetOffsetIfNeeded(TopicPartition partition, OffsetResetStrategy requestedResetStrategy, ListOffsetData offsetData) { 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.maybeSeek(partition, position.offset, requestedResetStrategy); } 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 cfd4929bbb26d..140830316e1fa 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 @@ -60,8 +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 thread-safe. The only exception is the consumer rebalance - * listener which is guaranteed to be executed without holding any locks. + * Thread Safety: this class is thread-safe. */ public class SubscriptionState { private static final String SUBSCRIPTION_EXCEPTION_MESSAGE = @@ -242,25 +241,24 @@ public synchronized boolean assignFromSubscribed(Collection assi 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( diff --git a/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java b/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java index 8913a4343c87e..22e183ab23865 100644 --- a/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java +++ b/clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java @@ -43,11 +43,11 @@ */ public class PartitionStates { - private final Map map = new LinkedHashMap<>(); + private final LinkedHashMap map = new LinkedHashMap<>(); private final Set partitionSetView = Collections.unmodifiableSet(map.keySet()); /* the number of partitions that are currently assigned available in a thread safe manner */ - private int size = 0; + private volatile int size = 0; public PartitionStates() {} diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml index c97eeeb68a817..721b05e464c0e 100644 --- a/gradle/spotbugs-exclude.xml +++ b/gradle/spotbugs-exclude.xml @@ -272,14 +272,6 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read - - - - - - From 0f089d0225286cae0b19911413d09a61f96de5b0 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Mon, 27 May 2019 11:34:49 +0200 Subject: [PATCH 5/7] Fixes after rebase --- .../kafka/clients/consumer/internals/SubscriptionState.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 140830316e1fa..0fef01e1ddce6 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 @@ -98,7 +98,7 @@ private enum SubscriptionType { private int assignmentId = 0; @Override - public String toString() { + public synchronized String toString() { return "SubscriptionState{" + "type=" + subscriptionType + ", subscribedPattern=" + subscribedPattern + @@ -108,7 +108,7 @@ public String toString() { ", assignment=" + assignment.partitionStateValues() + " (id=" + assignmentId + ")}"; } - public String prettyString() { + public synchronized String prettyString() { switch (subscriptionType) { case NONE: return "None"; From ae335050bd34d77bb95b7785f82228924872011d Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Tue, 28 May 2019 15:44:38 +0200 Subject: [PATCH 6/7] Fix test issues --- .../consumer/internals/FetcherTest.java | 36 +++++++++++++++---- 1 file changed, 29 insertions(+), 7 deletions(-) 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 9028e7fae08f8..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 @@ -1488,7 +1488,8 @@ public void testSeekWithInFlightReset() { @Test(timeout = 10000) public void testEarlierOffsetResetArrivesLate() throws InterruptedException { - buildFetcher(); + LogContext lc = new LogContext(); + buildFetcher(spy(new SubscriptionState(lc, OffsetResetStrategy.EARLIEST)), lc); assignFromUser(singleton(tp0)); ExecutorService es = Executors.newSingleThreadExecutor(); @@ -2863,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(), @@ -3621,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, @@ -3645,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 = spy(new SubscriptionState(logContext, offsetResetStrategy)); + subscriptions = subscriptionState; metadata = new ConsumerMetadata(0, metadataExpireMs, false, false, subscriptions, logContext, new ClusterResourceListeners()); client = new MockClient(time, metadata); From c6ef4274d4b3b39eeca4c68ffc7ae2520657f24f Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Wed, 29 May 2019 16:54:36 +0200 Subject: [PATCH 7/7] Address subscription related comments --- .../kafka/clients/consumer/internals/SubscriptionState.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 0fef01e1ddce6..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 @@ -181,7 +181,8 @@ private boolean changeSubscription(Set topicsToSubscribe) { return false; subscription = topicsToSubscribe; - groupSubscription = topicsToSubscribe; + groupSubscription = new HashSet<>(groupSubscription); + groupSubscription.addAll(topicsToSubscribe); return true; } @@ -201,8 +202,7 @@ synchronized boolean groupSubscribe(Collection topics) { * Reset the group's subscription to only contain topics subscribed by this consumer. */ synchronized void resetGroupSubscription() { - groupSubscription = new HashSet<>(groupSubscription); - groupSubscription.retainAll(subscription); + groupSubscription = subscription; } /**