Skip to content

KAFKA-7703: position() may return a wrong offset after seekToEnd #6407

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 7 commits into from
May 29, 2019
Merged

KAFKA-7703: position() may return a wrong offset after seekToEnd #6407

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
9acfc8c
KAFKA-7703: position() may return a wrong offset after "seekToEnd" is…
viktorsomogyi Mar 7, 2019
d2d4cca
Synchronize SubscriptionState instead of TopicPartitionState
viktorsomogyi Apr 25, 2019
f01800d
Synchronize preferred replica methods
viktorsomogyi May 21, 2019
818389b
Fix review comments
viktorsomogyi May 23, 2019
0f089d0
Fixes after rebase
viktorsomogyi May 27, 2019
ae33505
Fix test issues
viktorsomogyi May 28, 2019
c6ef427
Address subscription related comments
viktorsomogyi May 29, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 3 additions & 9 deletions clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -888,7 +888,7 @@ private static Metrics buildMetrics(ConsumerConfig config, Time time, String cli
public Set<TopicPartition> assignment() {
acquireAndEnsureOpen();
try {
return Collections.unmodifiableSet(new HashSet<>(this.subscriptions.assignedPartitions()));
return Collections.unmodifiableSet(this.subscriptions.assignedPartitions());
} finally {
release();
}
Expand Down Expand Up @@ -1605,10 +1605,7 @@ public void seekToBeginning(Collection<TopicPartition> partitions) {
acquireAndEnsureOpen();
try {
Collection<TopicPartition> 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();
}
Expand All @@ -1633,10 +1630,7 @@ public void seekToEnd(Collection<TopicPartition> partitions) {
acquireAndEnsureOpen();
try {
Collection<TopicPartition> 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();
}
Expand Down
8 changes: 3 additions & 5 deletions clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,7 @@ public synchronized ConsumerRecords<K, V> poll(final Duration timeout) {
public synchronized void addRecord(ConsumerRecord<K, V> record) {
ensureNotClosed();
TopicPartition tp = new TopicPartition(record.topic(), record.partition());
Set<TopicPartition> currentAssigned = new HashSet<>(this.subscriptions.assignedPartitions());
Set<TopicPartition> 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<ConsumerRecord<K, V>> recs = this.records.computeIfAbsent(tp, k -> new ArrayList<>());
Expand Down Expand Up @@ -312,8 +312,7 @@ public synchronized long position(TopicPartition partition, final Duration timeo
@Override
public synchronized void seekToBeginning(Collection<TopicPartition> partitions) {
ensureNotClosed();
for (TopicPartition tp : partitions)
subscriptions.requestOffsetReset(tp, OffsetResetStrategy.EARLIEST);
subscriptions.requestOffsetReset(partitions, OffsetResetStrategy.EARLIEST);
}

public synchronized void updateBeginningOffsets(Map<TopicPartition, Long> newOffsets) {
Expand All @@ -323,8 +322,7 @@ public synchronized void updateBeginningOffsets(Map<TopicPartition, Long> newOff
@Override
public synchronized void seekToEnd(Collection<TopicPartition> 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
Expand Down
5 changes: 2 additions & 3 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,7 +256,7 @@ protected void onJoinComplete(int generation,
return;
}

Set<TopicPartition> assignedPartitions = new HashSet<>(subscriptions.assignedPartitions());
Set<TopicPartition> 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.
Expand Down Expand Up @@ -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<TopicPartition> revoked = new HashSet<>(subscriptions.assignedPartitions());
Set<TopicPartition> revoked = subscriptions.assignedPartitions();
log.info("Revoking previously assigned partitions {}", revoked);
try {
listener.onPartitionsRevoked(revoked);
Expand Down
34 changes: 16 additions & 18 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
Expand Down Expand Up @@ -664,22 +673,11 @@ private List<ConsumerRecord<K, V>> 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<TopicPartition, Long> partitionResetTimestamps) {
Expand All @@ -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);
}
}

Expand Down Expand Up @@ -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<TopicPartition> newAssignedPartitions = new HashSet<>(subscription.assignedPartitions());
Set<TopicPartition> newAssignedPartitions = subscription.assignedPartitions();
for (TopicPartition tp : this.assignedPartitions) {
if (!newAssignedPartitions.contains(tp)) {
metrics.removeSensor(partitionLagMetricName(tp));
Expand Down
Loading