tchekai704
/
src-kafka
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

KAFKA-14966; [2/N] Extract OffsetFetcher reusable logic (#13898) 

This is a follow up on the initial OffsetFetcher refactoring to extract reusable logic, needed for the new consumer implementation (initial refactoring merged with PR-13815).

Similar to the initial refactoring, this PR brings no changes to the existing logic, just extracting functions or pieces of logic.

There were no individual tests for the extracted functions, so no tests were migrated.

Reviewers: Jun Rao <junrao@gmail.com>
pull/13967/head
Lianet Magrans 1 year ago committed by GitHub
parent
98fbd8afc7
commit
4a61b48d3d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 98 additions and 75 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 85
      clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java
  2. 86
      clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java
  3. 2
      clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

85
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java
Unescape View File

@ -24,7 +24,6 @@ import org.apache.kafka.clients.StaleMetadataException; @@ -24,7 +24,6 @@ import org.apache.kafka.clients.StaleMetadataException;
import org.apache.kafka.clients.consumer.LogTruncationException;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.ListOffsetData;
import org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.ListOffsetResult;
import org.apache.kafka.clients.consumer.internals.OffsetsForLeaderEpochClient.OffsetForEpochResult;
@ -34,12 +33,9 @@ import org.apache.kafka.common.Node; @@ -34,12 +33,9 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.ListOffsetsRequest;
import org.apache.kafka.common.requests.ListOffsetsResponse;
import org.apache.kafka.common.requests.OffsetsForLeaderEpochRequest;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Timer;
@ -54,10 +50,14 @@ import java.util.Map; @@ -54,10 +50,14 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.buildOffsetsForTimesResult;
import static org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.hasUsableOffsetForLeaderEpochVersion;
import static org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.regroupFetchPositionsByLeader;
import static org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.topicsForPartitions;
/**
* {@link OffsetFetcher} is responsible for fetching the {@link OffsetAndTimestamp offsets} for
* a given set of {@link TopicPartition topic and partition pairs} and for validation and resetting of positions,
@ -73,7 +73,6 @@ public class OffsetFetcher { @@ -73,7 +73,6 @@ public class OffsetFetcher {
private final long retryBackoffMs;
private final long requestTimeoutMs;
private final IsolationLevel isolationLevel;
private final AtomicReference<RuntimeException> cachedListOffsetsException = new AtomicReference<>();
private final OffsetsForLeaderEpochClient offsetsForLeaderEpochClient;
private final ApiVersions apiVersions;
private final OffsetFetcherUtils offsetFetcherUtils;
@ -98,16 +97,7 @@ public class OffsetFetcher { @@ -98,16 +97,7 @@ public class OffsetFetcher {
this.apiVersions = apiVersions;
this.offsetsForLeaderEpochClient = new OffsetsForLeaderEpochClient(client, logContext);
this.offsetFetcherUtils = new OffsetFetcherUtils(logContext, metadata, subscriptions,
time, apiVersions);
}
private OffsetResetStrategy timestampToOffsetResetStrategy(long timestamp) {
if (timestamp == ListOffsetsRequest.EARLIEST_TIMESTAMP)
return OffsetResetStrategy.EARLIEST;
else if (timestamp == ListOffsetsRequest.LATEST_TIMESTAMP)
return OffsetResetStrategy.LATEST;
else
return null;
time, retryBackoffMs, apiVersions);
}
/**
@ -117,11 +107,6 @@ public class OffsetFetcher { @@ -117,11 +107,6 @@ public class OffsetFetcher {
* and one or more partitions aren't awaiting a seekToBeginning() or seekToEnd().
*/
public void resetPositionsIfNeeded() {
// Raise exception from previous offset fetch if there is one
RuntimeException exception = cachedListOffsetsException.getAndSet(null);
if (exception != null)
throw exception;
Map<TopicPartition, Long> offsetResetTimestamps = offsetFetcherUtils.getOffsetResetTimestamp();
if (offsetResetTimestamps.isEmpty())
@ -142,13 +127,13 @@ public class OffsetFetcher { @@ -142,13 +127,13 @@ public class OffsetFetcher {
public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch,
Timer timer) {
metadata.addTransientTopics(offsetFetcherUtils.topicsForPartitions(timestampsToSearch.keySet()));
metadata.addTransientTopics(topicsForPartitions(timestampsToSearch.keySet()));
try {
Map<TopicPartition, ListOffsetData> fetchedOffsets = fetchOffsetsByTimes(timestampsToSearch,
timer, true).fetchedOffsets;
return offsetFetcherUtils.buildOffsetsForTimesResult(timestampsToSearch, fetchedOffsets);
return buildOffsetsForTimesResult(timestampsToSearch, fetchedOffsets);
} finally {
metadata.clearTransientTopics();
}
@ -215,7 +200,7 @@ public class OffsetFetcher { @@ -215,7 +200,7 @@ public class OffsetFetcher {
private Map<TopicPartition, Long> beginningOrEndOffset(Collection<TopicPartition> partitions,
long timestamp,
Timer timer) {
metadata.addTransientTopics(offsetFetcherUtils.topicsForPartitions(partitions));
metadata.addTransientTopics(topicsForPartitions(partitions));
try {
Map<TopicPartition, Long> timestampsToSearch = partitions.stream()
.distinct()
@ -230,16 +215,6 @@ public class OffsetFetcher { @@ -230,16 +215,6 @@ public class OffsetFetcher {
}
}
// Visible for testing
void resetPositionIfNeeded(TopicPartition partition, OffsetResetStrategy requestedResetStrategy, ListOffsetData offsetData) {
FetchPosition position = new FetchPosition(
offsetData.offset,
Optional.empty(), // This will ensure we skip validation
metadata.currentLeader(partition));
offsetData.leaderEpoch.ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(partition, epoch));
subscriptions.maybeSeekUnvalidated(partition, position, requestedResetStrategy);
}
private void resetPositionsAsync(Map<TopicPartition, Long> partitionResetTimestamps) {
Map<Node, Map<TopicPartition, ListOffsetsPartition>> timestampsToSearchByNode =
groupListOffsetRequests(partitionResetTimestamps, new HashSet<>());
@ -252,39 +227,17 @@ public class OffsetFetcher { @@ -252,39 +227,17 @@ public class OffsetFetcher {
future.addListener(new RequestFutureListener<ListOffsetResult>() {
@Override
public void onSuccess(ListOffsetResult result) {
if (!result.partitionsToRetry.isEmpty()) {
subscriptions.requestFailed(result.partitionsToRetry, time.milliseconds() + retryBackoffMs);
metadata.requestUpdate();
}
for (Map.Entry<TopicPartition, ListOffsetData> fetchedOffset : result.fetchedOffsets.entrySet()) {
TopicPartition partition = fetchedOffset.getKey();
ListOffsetData offsetData = fetchedOffset.getValue();
ListOffsetsPartition requestedReset = resetTimestamps.get(partition);
resetPositionIfNeeded(partition, timestampToOffsetResetStrategy(requestedReset.timestamp()), offsetData);
}
offsetFetcherUtils.onSuccessfulRequestForResettingPositions(resetTimestamps, result);
}
@Override
public void onFailure(RuntimeException e) {
subscriptions.requestFailed(resetTimestamps.keySet(), time.milliseconds() + retryBackoffMs);
metadata.requestUpdate();
if (!(e instanceof RetriableException) && !cachedListOffsetsException.compareAndSet(null, e))
log.error("Discarding error in ListOffsetResponse because another error is pending", e);
offsetFetcherUtils.onFailedRequestForResettingPositions(resetTimestamps, e);
}
});
}
}
static boolean hasUsableOffsetForLeaderEpochVersion(NodeApiVersions nodeApiVersions) {
ApiVersion apiVersion = nodeApiVersions.apiVersion(ApiKeys.OFFSET_FOR_LEADER_EPOCH);
if (apiVersion == null)
return false;
return OffsetsForLeaderEpochRequest.supportsTopicPermission(apiVersion.maxVersion());
}
/**
* For each partition which needs validation, make an asynchronous request to get the end-offsets for the partition
* with the epoch less than or equal to the epoch the partition last saw.
@ -294,8 +247,7 @@ public class OffsetFetcher { @@ -294,8 +247,7 @@ public class OffsetFetcher {
* Requests are grouped by Node for efficiency.
*/
private void validatePositionsAsync(Map<TopicPartition, FetchPosition> partitionsToValidate) {
final Map<Node, Map<TopicPartition, FetchPosition>> regrouped =
regroupFetchPositionsByLeader(partitionsToValidate);
final Map<Node, Map<TopicPartition, FetchPosition>> regrouped = regroupFetchPositionsByLeader(partitionsToValidate);
long nextResetTimeMs = time.milliseconds() + requestTimeoutMs;
regrouped.forEach((node, fetchPositions) -> {
@ -518,7 +470,6 @@ public class OffsetFetcher { @@ -518,7 +470,6 @@ public class OffsetFetcher {
}
}
/**
* If we have seen new metadata (as tracked by {@link org.apache.kafka.clients.Metadata#updateVersion()}), then
* we should check that all the assignments have a valid position.
@ -526,14 +477,4 @@ public class OffsetFetcher { @@ -526,14 +477,4 @@ public class OffsetFetcher {
public void validatePositionsOnMetadataChange() {
offsetFetcherUtils.validatePositionsOnMetadataChange();
}
private Map<Node, Map<TopicPartition, FetchPosition>> regroupFetchPositionsByLeader(
Map<TopicPartition, FetchPosition> partitionMap) {
return partitionMap.entrySet()
.stream()
.filter(entry -> entry.getValue().currentLeader.leader.isPresent())
.collect(Collectors.groupingBy(entry -> entry.getValue().currentLeader.leader.get(),
Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
}
}
}

86
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java
Unescape View File

@ -17,16 +17,22 @@ @@ -17,16 +17,22 @@
package org.apache.kafka.clients.consumer.internals;
import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.message.ApiVersionsResponseData;
import org.apache.kafka.common.message.ListOffsetsRequestData;
import org.apache.kafka.common.message.ListOffsetsResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.ListOffsetsRequest;
import org.apache.kafka.common.requests.ListOffsetsResponse;
import org.apache.kafka.common.requests.OffsetsForLeaderEpochRequest;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
@ -49,21 +55,25 @@ class OffsetFetcherUtils { @@ -49,21 +55,25 @@ class OffsetFetcherUtils {
private final ConsumerMetadata metadata;
private final SubscriptionState subscriptionState;
private final Time time;
private final long retryBackoffMs;
private final ApiVersions apiVersions;
private final Logger log;
private final AtomicReference<RuntimeException> cachedOffsetForLeaderException = new AtomicReference<>();
private final AtomicReference<RuntimeException> cachedListOffsetsException = new AtomicReference<>();
private final AtomicInteger metadataUpdateVersion = new AtomicInteger(-1);
OffsetFetcherUtils(LogContext logContext,
ConsumerMetadata metadata,
SubscriptionState subscriptionState,
Time time,
long retryBackoffMs,
ApiVersions apiVersions) {
this.log = logContext.logger(getClass());
this.metadata = metadata;
this.subscriptionState = subscriptionState;
this.time = time;
this.retryBackoffMs = retryBackoffMs;
this.apiVersions = apiVersions;
}
@ -198,6 +208,11 @@ class OffsetFetcherUtils { @@ -198,6 +208,11 @@ class OffsetFetcherUtils {
}
Map<TopicPartition, Long> getOffsetResetTimestamp() {
// Raise exception from previous offset fetch if there is one
RuntimeException exception = cachedListOffsetsException.getAndSet(null);
if (exception != null)
throw exception;
Set<TopicPartition> partitions = subscriptionState.partitionsNeedingReset(time.milliseconds());
final Map<TopicPartition, Long> offsetResetTimestamps = new HashMap<>();
for (final TopicPartition partition : partitions) {
@ -209,7 +224,7 @@ class OffsetFetcherUtils { @@ -209,7 +224,7 @@ class OffsetFetcherUtils {
return offsetResetTimestamps;
}
Map<TopicPartition, OffsetAndTimestamp> buildOffsetsForTimesResult(final Map<TopicPartition, Long> timestampsToSearch,
static Map<TopicPartition, OffsetAndTimestamp> buildOffsetsForTimesResult(final Map<TopicPartition, Long> timestampsToSearch,
final Map<TopicPartition, ListOffsetData> fetchedOffsets) {
HashMap<TopicPartition, OffsetAndTimestamp> offsetsByTimes = new HashMap<>(timestampsToSearch.size());
for (Map.Entry<TopicPartition, Long> entry : timestampsToSearch.entrySet())
@ -235,7 +250,7 @@ class OffsetFetcherUtils { @@ -235,7 +250,7 @@ class OffsetFetcherUtils {
return null;
}
Set<String> topicsForPartitions(Collection<TopicPartition> partitions) {
static Set<String> topicsForPartitions(Collection<TopicPartition> partitions) {
return partitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
}
@ -261,6 +276,73 @@ class OffsetFetcherUtils { @@ -261,6 +276,73 @@ class OffsetFetcherUtils {
}
}
static OffsetResetStrategy timestampToOffsetResetStrategy(long timestamp) {
if (timestamp == ListOffsetsRequest.EARLIEST_TIMESTAMP)
return OffsetResetStrategy.EARLIEST;
else if (timestamp == ListOffsetsRequest.LATEST_TIMESTAMP)
return OffsetResetStrategy.LATEST;
else
return null;
}
void onSuccessfulRequestForResettingPositions(
final Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition> resetTimestamps,
final ListOffsetResult result) {
if (!result.partitionsToRetry.isEmpty()) {
subscriptionState.requestFailed(result.partitionsToRetry, time.milliseconds() + retryBackoffMs);
metadata.requestUpdate();
}
for (Map.Entry<TopicPartition, ListOffsetData> fetchedOffset : result.fetchedOffsets.entrySet()) {
TopicPartition partition = fetchedOffset.getKey();
ListOffsetData offsetData = fetchedOffset.getValue();
ListOffsetsRequestData.ListOffsetsPartition requestedReset = resetTimestamps.get(partition);
resetPositionIfNeeded(
partition,
timestampToOffsetResetStrategy(requestedReset.timestamp()),
offsetData);
}
}
void onFailedRequestForResettingPositions(
final Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition> resetTimestamps,
final RuntimeException error) {
subscriptionState.requestFailed(resetTimestamps.keySet(), time.milliseconds() + retryBackoffMs);
metadata.requestUpdate();
if (!(error instanceof RetriableException) && !cachedListOffsetsException.compareAndSet(null,
error))
log.error("Discarding error in ListOffsetResponse because another error is pending", error);
}
// Visible for testing
void resetPositionIfNeeded(TopicPartition partition, OffsetResetStrategy requestedResetStrategy,
ListOffsetData offsetData) {
SubscriptionState.FetchPosition position = new SubscriptionState.FetchPosition(
offsetData.offset,
Optional.empty(), // This will ensure we skip validation
metadata.currentLeader(partition));
offsetData.leaderEpoch.ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(partition, epoch));
subscriptionState.maybeSeekUnvalidated(partition, position, requestedResetStrategy);
}
static Map<Node, Map<TopicPartition, SubscriptionState.FetchPosition>> regroupFetchPositionsByLeader(
Map<TopicPartition, SubscriptionState.FetchPosition> partitionMap) {
return partitionMap.entrySet()
.stream()
.filter(entry -> entry.getValue().currentLeader.leader.isPresent())
.collect(Collectors.groupingBy(entry -> entry.getValue().currentLeader.leader.get(),
Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
}
static boolean hasUsableOffsetForLeaderEpochVersion(NodeApiVersions nodeApiVersions) {
ApiVersionsResponseData.ApiVersion apiVersion = nodeApiVersions.apiVersion(ApiKeys.OFFSET_FOR_LEADER_EPOCH);
if (apiVersion == null)
return false;
return OffsetsForLeaderEpochRequest.supportsTopicPermission(apiVersion.maxVersion());
}
static class ListOffsetResult {
final Map<TopicPartition, OffsetFetcherUtils.ListOffsetData> fetchedOffsets;
final Set<TopicPartition> partitionsToRetry;

2
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
Unescape View File

@ -47,7 +47,7 @@ import java.util.function.LongSupplier; @@ -47,7 +47,7 @@ import java.util.function.LongSupplier;
import java.util.function.Predicate;
import java.util.regex.Pattern;
import static org.apache.kafka.clients.consumer.internals.OffsetFetcher.hasUsableOffsetForLeaderEpochVersion;
import static org.apache.kafka.clients.consumer.internals.OffsetFetcherUtils.hasUsableOffsetForLeaderEpochVersion;
import static org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.UNDEFINED_EPOCH;
import static org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.UNDEFINED_EPOCH_OFFSET;

Write Preview
Loading…
Cancel
Save