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 c33a52e396109..7f63e8bbc65d7 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 @@ -744,6 +744,7 @@ else if (enableAutoCommit) Sensor throttleTimeSensor = Fetcher.throttleTimeSensor(metrics, metricsRegistry); int heartbeatIntervalMs = config.getInt(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG); + ApiVersions apiVersions = new ApiVersions(); NetworkClient netClient = new NetworkClient( new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, metricGrpPrefix, channelBuilder, logContext), this.metadata, @@ -757,7 +758,7 @@ else if (enableAutoCommit) ClientDnsLookup.forConfig(config.getString(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG)), time, true, - new ApiVersions(), + apiVersions, throttleTimeSensor, logContext); this.client = new ConsumerNetworkClient( @@ -813,7 +814,8 @@ else if (enableAutoCommit) this.time, this.retryBackoffMs, this.requestTimeoutMs, - isolationLevel); + isolationLevel, + apiVersions); config.logUnused(); AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds()); 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 7b633d684e5ba..aa2936a3c1147 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 @@ -16,10 +16,12 @@ */ package org.apache.kafka.clients.consumer.internals; +import org.apache.kafka.clients.ApiVersions; import org.apache.kafka.clients.ClientResponse; import org.apache.kafka.clients.FetchSessionHandler; import org.apache.kafka.clients.Metadata; import org.apache.kafka.clients.MetadataCache; +import org.apache.kafka.clients.NodeApiVersions; import org.apache.kafka.clients.StaleMetadataException; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -49,6 +51,7 @@ import org.apache.kafka.common.metrics.stats.Meter; import org.apache.kafka.common.metrics.stats.Min; import org.apache.kafka.common.metrics.stats.Value; +import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.record.BufferSupplier; import org.apache.kafka.common.record.ControlRecordType; @@ -57,6 +60,7 @@ import org.apache.kafka.common.record.RecordBatch; import org.apache.kafka.common.record.Records; import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.common.requests.ApiVersionsResponse; import org.apache.kafka.common.requests.FetchRequest; import org.apache.kafka.common.requests.FetchResponse; import org.apache.kafka.common.requests.IsolationLevel; @@ -64,6 +68,7 @@ import org.apache.kafka.common.requests.ListOffsetResponse; import org.apache.kafka.common.requests.MetadataRequest; import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.requests.OffsetsForLeaderEpochRequest; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.utils.CloseableIterator; import org.apache.kafka.common.utils.LogContext; @@ -144,6 +149,7 @@ public class Fetcher implements Closeable { private final AtomicReference cachedOffsetForLeaderException = new AtomicReference<>(); private final OffsetsForLeaderEpochClient offsetsForLeaderEpochClient; private final Set nodesWithPendingFetchRequests; + private final ApiVersions apiVersions; private PartitionRecords nextInLineRecords = null; @@ -165,7 +171,8 @@ public Fetcher(LogContext logContext, Time time, long retryBackoffMs, long requestTimeoutMs, - IsolationLevel isolationLevel) { + IsolationLevel isolationLevel, + ApiVersions apiVersions) { this.log = logContext.logger(Fetcher.class); this.logContext = logContext; this.time = time; @@ -186,6 +193,7 @@ public Fetcher(LogContext logContext, this.retryBackoffMs = retryBackoffMs; this.requestTimeoutMs = requestTimeoutMs; this.isolationLevel = isolationLevel; + this.apiVersions = apiVersions; this.sessionHandlers = new HashMap<>(); this.offsetsForLeaderEpochClient = new OffsetsForLeaderEpochClient(client, logContext); this.nodesWithPendingFetchRequests = new HashSet<>(); @@ -711,6 +719,19 @@ public void onFailure(RuntimeException e) { } } + private boolean hasUsableOffsetForLeaderEpochVersion(Node node) { + NodeApiVersions nodeApiVersions = apiVersions.get(node.idString()); + + if (nodeApiVersions == null) + return false; + + ApiVersionsResponse.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. @@ -727,12 +748,22 @@ private void validateOffsetsAsync(Map cachedLeaderAndEpochs = partitionsToValidate.entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().currentLeader)); + if (!hasUsableOffsetForLeaderEpochVersion(node)) { + log.debug("Skipping validation of fetch offsets for partitions {} since the broker does not " + + "support the required protocol version (introduced in Kafka 2.3)", + cachedLeaderAndEpochs.keySet()); + for (TopicPartition partition : cachedLeaderAndEpochs.keySet()) { + subscriptions.completeValidation(partition); + } + return; + } + + subscriptions.setNextAllowedRetry(dataMap.keySet(), time.milliseconds() + requestTimeoutMs); + RequestFuture future = offsetsForLeaderEpochClient.sendAsyncRequest(node, partitionsToValidate); future.addListener(new RequestFutureListener() { @Override @@ -772,7 +803,7 @@ public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsRe } } else { // Offset is fine, clear the validation state - subscriptions.validate(respTopicPartition); + subscriptions.completeValidation(respTopicPartition); } } }); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsForLeaderEpochClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsForLeaderEpochClient.java index d7b02a791ac98..9556d3c64dc1b 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsForLeaderEpochClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsForLeaderEpochClient.java @@ -19,7 +19,6 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.TopicAuthorizationException; -import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.requests.AbstractRequest; import org.apache.kafka.common.requests.EpochEndOffset; @@ -53,7 +52,7 @@ protected AbstractRequest.Builder prepareRequest( fetchEpoch -> partitionData.put(topicPartition, new OffsetsForLeaderEpochRequest.PartitionData(fetchPosition.currentLeader.epoch, fetchEpoch)))); - return OffsetsForLeaderEpochRequest.Builder.forConsumer(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(), partitionData); + return OffsetsForLeaderEpochRequest.Builder.forConsumer(partitionData); } @Override 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 c48bd741ad7fa..4c87dba38204a 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 @@ -408,7 +408,7 @@ public boolean awaitingValidation(TopicPartition tp) { return assignedState(tp).awaitingValidation(); } - public void validate(TopicPartition tp) { + public void completeValidation(TopicPartition tp) { assignedState(tp).validate(); } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java index a37b6da0b7273..b3846d1bb3aa0 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java @@ -84,6 +84,10 @@ public ApiVersion(ApiKeys apiKey) { this(apiKey.id, apiKey.oldestVersion(), apiKey.latestVersion()); } + public ApiVersion(ApiKeys apiKey, short minVersion, short maxVersion) { + this(apiKey.id, minVersion, maxVersion); + } + public ApiVersion(short apiKey, short minVersion, short maxVersion) { this.apiKey = apiKey; this.minVersion = minVersion; diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java index d3df6cfba2247..d5c78da5e281f 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java @@ -17,6 +17,7 @@ package org.apache.kafka.common.requests; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.protocol.types.Field; @@ -44,6 +45,11 @@ public class OffsetsForLeaderEpochRequest extends AbstractRequest { */ public static final int CONSUMER_REPLICA_ID = -1; + /** + * Sentinel replica_id which indicates either a debug consumer or a replica which is using + * an old version of the protocol. + */ + public static final int DEBUGGING_REPLICA_ID = -2; private static final Field.ComplexArray TOPICS = new Field.ComplexArray("topics", "An array of topics to get epochs for"); @@ -101,23 +107,29 @@ public static class Builder extends AbstractRequest.Builder epochsByPartition; private final int replicaId; - Builder(short version, Map epochsByPartition, int replicaId) { - super(ApiKeys.OFFSET_FOR_LEADER_EPOCH, version); + Builder(short oldestAllowedVersion, short latestAllowedVersion, Map epochsByPartition, int replicaId) { + super(ApiKeys.OFFSET_FOR_LEADER_EPOCH, oldestAllowedVersion, latestAllowedVersion); this.epochsByPartition = epochsByPartition; this.replicaId = replicaId; } - public static Builder forConsumer(short version, Map epochsByPartition) { - return new Builder(version, epochsByPartition, CONSUMER_REPLICA_ID); + public static Builder forConsumer(Map epochsByPartition) { + // Old versions of this API require CLUSTER permission which is not typically granted + // to clients. Beginning with version 3, the broker requires only TOPIC Describe + // permission for the topic of each requested partition. In order to ensure client + // compatibility, we only send this request when we can guarantee the relaxed permissions. + return new Builder((short) 3, ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(), + epochsByPartition, CONSUMER_REPLICA_ID); } public static Builder forFollower(short version, Map epochsByPartition, int replicaId) { - return new Builder(version, epochsByPartition, replicaId); - + return new Builder(version, version, epochsByPartition, replicaId); } @Override public OffsetsForLeaderEpochRequest build(short version) { + if (version < oldestAllowedVersion() || version > latestAllowedVersion()) + throw new UnsupportedVersionException("Cannot build " + this + " with version " + version); return new OffsetsForLeaderEpochRequest(epochsByPartition, replicaId, version); } @@ -143,7 +155,7 @@ public OffsetsForLeaderEpochRequest(Map epochsByP public OffsetsForLeaderEpochRequest(Struct struct, short version) { super(ApiKeys.OFFSET_FOR_LEADER_EPOCH, version); - replicaId = struct.getOrElse(REPLICA_ID, CONSUMER_REPLICA_ID); + replicaId = struct.getOrElse(REPLICA_ID, DEBUGGING_REPLICA_ID); epochsByPartition = new HashMap<>(); for (Object topicAndEpochsObj : struct.get(TOPICS)) { Struct topicAndEpochs = (Struct) topicAndEpochsObj; @@ -222,4 +234,13 @@ public String toString() { return bld.toString(); } } + + /** + * Check whether a broker allows Topic-level permissions in order to use the + * OffsetForLeaderEpoch API. Old versions require Cluster permission. + */ + public static boolean supportsTopicPermission(short latestUsableVersion) { + return latestUsableVersion >= 3; + } + } diff --git a/clients/src/main/resources/common/message/OffsetForLeaderEpochRequest.json b/clients/src/main/resources/common/message/OffsetForLeaderEpochRequest.json index 84585ca384ded..53bef12d7ebf9 100644 --- a/clients/src/main/resources/common/message/OffsetForLeaderEpochRequest.json +++ b/clients/src/main/resources/common/message/OffsetForLeaderEpochRequest.json @@ -18,12 +18,13 @@ "type": "request", "name": "OffsetForLeaderEpochRequest", // Version 1 is the same as version 0. - // // Version 2 adds the current leader epoch to support fencing. - // Version 3 adds ReplicaId + // Version 3 adds ReplicaId (the default is -2 which conventionally represents a + // "debug" consumer which is allowed to see offsets beyond the high watermark). + // Followers will use this replicaId when using an older version of the protocol. "validVersions": "0-3", "fields": [ - { "name": "ReplicaId", "type": "int32", "versions": "3+", + { "name": "ReplicaId", "type": "int32", "versions": "3+", "default": -2, "ignorable": true, "about": "The broker ID of the follower, of -1 if this request is from a consumer." }, { "name": "Topics", "type": "[]OffsetForLeaderTopic", "versions": "0+", "about": "Each topic to get offsets for.", "fields": [ diff --git a/clients/src/test/java/org/apache/kafka/clients/ApiVersionsTest.java b/clients/src/test/java/org/apache/kafka/clients/ApiVersionsTest.java index 654a6064d5401..a66a86fb959b7 100644 --- a/clients/src/test/java/org/apache/kafka/clients/ApiVersionsTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/ApiVersionsTest.java @@ -36,7 +36,7 @@ public void testMaxUsableProduceMagic() { assertEquals(RecordBatch.CURRENT_MAGIC_VALUE, apiVersions.maxUsableProduceMagic()); apiVersions.update("1", NodeApiVersions.create(Collections.singleton( - new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE.id, (short) 0, (short) 2)))); + new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE, (short) 0, (short) 2)))); assertEquals(RecordBatch.MAGIC_VALUE_V1, apiVersions.maxUsableProduceMagic()); apiVersions.remove("1"); diff --git a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java index db5c7df8742b0..b5bba716296b0 100644 --- a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java @@ -272,12 +272,11 @@ public void testConnectionThrottling() { // Creates expected ApiVersionsResponse from the specified node, where the max protocol version for the specified // key is set to the specified version. - private ApiVersionsResponse createExpectedApiVersionsResponse(Node node, ApiKeys key, - short apiVersionsMaxProtocolVersion) { + private ApiVersionsResponse createExpectedApiVersionsResponse(ApiKeys key, short maxVersion) { List versionList = new ArrayList<>(); for (ApiKeys apiKey : ApiKeys.values()) { if (apiKey == key) { - versionList.add(new ApiVersionsResponse.ApiVersion(apiKey.id, (short) 0, apiVersionsMaxProtocolVersion)); + versionList.add(new ApiVersionsResponse.ApiVersion(apiKey, (short) 0, maxVersion)); } else { versionList.add(new ApiVersionsResponse.ApiVersion(apiKey)); } @@ -289,7 +288,7 @@ private ApiVersionsResponse createExpectedApiVersionsResponse(Node node, ApiKeys public void testThrottlingNotEnabledForConnectionToOlderBroker() { // Instrument the test so that the max protocol version for PRODUCE returned from the node is 5 and thus // client-side throttling is not enabled. Also, return a response with a 100ms throttle delay. - setExpectedApiVersionsResponse(createExpectedApiVersionsResponse(node, ApiKeys.PRODUCE, (short) 5)); + setExpectedApiVersionsResponse(createExpectedApiVersionsResponse(ApiKeys.PRODUCE, (short) 5)); while (!client.ready(node, time.milliseconds())) client.poll(1, time.milliseconds()); selector.clear(); diff --git a/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java b/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java index 88d0c2e719e2a..58fb363ad5378 100644 --- a/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java @@ -58,7 +58,7 @@ public void testVersionsToString() { List versionList = new ArrayList<>(); for (ApiKeys apiKey : ApiKeys.values()) { if (apiKey == ApiKeys.DELETE_TOPICS) { - versionList.add(new ApiVersion(apiKey.id, (short) 10000, (short) 10001)); + versionList.add(new ApiVersion(apiKey, (short) 10000, (short) 10001)); } else { versionList.add(new ApiVersion(apiKey)); } @@ -93,7 +93,7 @@ public void testVersionsToString() { @Test public void testLatestUsableVersion() { NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton( - new ApiVersion(ApiKeys.PRODUCE.id, (short) 1, (short) 3))); + new ApiVersion(ApiKeys.PRODUCE, (short) 1, (short) 3))); assertEquals(3, apiVersions.latestUsableVersion(ApiKeys.PRODUCE)); assertEquals(1, apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 0, (short) 1)); assertEquals(1, apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 1, (short) 1)); @@ -108,14 +108,14 @@ public void testLatestUsableVersion() { @Test(expected = UnsupportedVersionException.class) public void testLatestUsableVersionOutOfRangeLow() { NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton( - new ApiVersion(ApiKeys.PRODUCE.id, (short) 1, (short) 2))); + new ApiVersion(ApiKeys.PRODUCE, (short) 1, (short) 2))); apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 3, (short) 4); } @Test(expected = UnsupportedVersionException.class) public void testLatestUsableVersionOutOfRangeHigh() { NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton( - new ApiVersion(ApiKeys.PRODUCE.id, (short) 2, (short) 3))); + new ApiVersion(ApiKeys.PRODUCE, (short) 2, (short) 3))); apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 0, (short) 1); } @@ -129,7 +129,7 @@ public void testUsableVersionCalculationNoKnownVersions() { @Test(expected = UnsupportedVersionException.class) public void testLatestUsableVersionOutOfRange() { NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton( - new ApiVersion(ApiKeys.PRODUCE.id, (short) 300, (short) 300))); + new ApiVersion(ApiKeys.PRODUCE, (short) 300, (short) 300))); apiVersions.latestUsableVersion(ApiKeys.PRODUCE); } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index 42cccd49a9be3..8e6bd01716a5b 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.clients.consumer; +import org.apache.kafka.clients.ApiVersions; import org.apache.kafka.clients.ClientRequest; import org.apache.kafka.clients.KafkaClient; import org.apache.kafka.clients.MockClient; @@ -1924,7 +1925,8 @@ private KafkaConsumer newConsumer(Time time, time, retryBackoffMs, requestTimeoutMs, - IsolationLevel.READ_UNCOMMITTED); + IsolationLevel.READ_UNCOMMITTED, + new ApiVersions()); return new KafkaConsumer<>( loggerFactory, 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 0e2662a82fa5a..69036cafc5bb3 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 @@ -157,6 +157,7 @@ public class FetcherTest { private FetcherMetricsRegistry metricsRegistry; private MockClient client; private Metrics metrics; + private ApiVersions apiVersions = new ApiVersions(); private ConsumerNetworkClient consumerClient; private Fetcher fetcher; @@ -778,7 +779,7 @@ public void testFetchRequestWhenRecordTooLarge() { buildFetcher(); client.setNodeApiVersions(NodeApiVersions.create(Collections.singletonList( - new ApiVersionsResponse.ApiVersion(ApiKeys.FETCH.id, (short) 2, (short) 2)))); + new ApiVersionsResponse.ApiVersion(ApiKeys.FETCH, (short) 2, (short) 2)))); makeFetchRequestWithIncompleteRecord(); try { fetcher.fetchedRecords(); @@ -2836,7 +2837,8 @@ public void testFetcherConcurrency() throws Exception { time, retryBackoffMs, requestTimeoutMs, - IsolationLevel.READ_UNCOMMITTED) { + IsolationLevel.READ_UNCOMMITTED, + apiVersions) { @Override protected FetchSessionHandler sessionHandler(int id) { final FetchSessionHandler handler = super.sessionHandler(id); @@ -3206,6 +3208,43 @@ public void testSubscriptionPositionUpdatedWithEpoch() { assertOptional(subscriptions.position(tp0).offsetEpoch, value -> assertEquals(value.intValue(), 1)); } + @Test + public void testOffsetValidationSkippedForOldBroker() { + // Old brokers may require CLUSTER permission to use the OffsetForLeaderEpoch API, + // so we should skip offset validation and not send the request. + + buildFetcher(); + assignFromUser(singleton(tp0)); + + Map partitionCounts = new HashMap<>(); + partitionCounts.put(tp0.topic(), 4); + + final int epochOne = 1; + final int epochTwo = 2; + + // Start with metadata, epoch=1 + metadata.update(TestUtils.metadataUpdateWith("dummy", 1, + Collections.emptyMap(), partitionCounts, tp -> epochOne), 0L); + + // Offset validation requires OffsetForLeaderEpoch request v3 or higher + Node node = metadata.fetch().nodes().get(0); + apiVersions.update(node.idString(), NodeApiVersions.create(singleton( + new ApiVersionsResponse.ApiVersion(ApiKeys.OFFSET_FOR_LEADER_EPOCH, (short) 0, (short) 2)))); + + // Seek with a position and leader+epoch + Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch( + metadata.leaderAndEpoch(tp0).leader, Optional.of(epochOne)); + subscriptions.seek(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch)); + + // Update metadata to epoch=2, enter validation + metadata.update(TestUtils.metadataUpdateWith("dummy", 1, + Collections.emptyMap(), partitionCounts, tp -> epochTwo), 0L); + fetcher.validateOffsetsIfNeeded(); + + // Offset validation is skipped + assertFalse(subscriptions.awaitingValidation(tp0)); + } + @Test public void testOffsetValidationFencing() { buildFetcher(); @@ -3221,6 +3260,10 @@ public void testOffsetValidationFencing() { // Start with metadata, epoch=1 metadata.update(TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, tp -> epochOne), 0L); + // Offset validation requires OffsetForLeaderEpoch request v3 or higher + Node node = metadata.fetch().nodes().get(0); + apiVersions.update(node.idString(), NodeApiVersions.create()); + // Seek with a position and leader+epoch Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(metadata.leaderAndEpoch(tp0).leader, Optional.of(epochOne)); subscriptions.seek(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch)); @@ -3231,7 +3274,7 @@ public void testOffsetValidationFencing() { assertTrue(subscriptions.awaitingValidation(tp0)); // Update the position to epoch=3, as we would from a fetch - subscriptions.validate(tp0); + subscriptions.completeValidation(tp0); SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition( 10, Optional.of(epochTwo), @@ -3287,6 +3330,10 @@ public void testTruncationDetected() { MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, tp -> 2); metadata.update(metadataResponse, 0L); + // Offset validation requires OffsetForLeaderEpoch request v3 or higher + Node node = metadata.fetch().nodes().get(0); + apiVersions.update(node.idString(), NodeApiVersions.create()); + // Seek Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(metadata.leaderAndEpoch(tp0).leader, Optional.of(1)); subscriptions.seek(tp0, new SubscriptionState.FetchPosition(0, Optional.of(1), leaderAndEpoch)); @@ -3547,7 +3594,8 @@ private void buildFetcher(MetricConfig metricConfig, time, retryBackoffMs, requestTimeoutMs, - isolationLevel); + isolationLevel, + apiVersions); } diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java index 5061447b54499..a58f3b5977bef 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java @@ -188,7 +188,7 @@ private void testAppendLargeOldMessageFormat(CompressionType compressionType) th ApiVersions apiVersions = new ApiVersions(); apiVersions.update(node1.idString(), NodeApiVersions.create(Collections.singleton( - new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE.id, (short) 0, (short) 2)))); + new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE, (short) 0, (short) 2)))); RecordAccumulator accum = createTestRecordAccumulator( batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, compressionType, 0); @@ -705,7 +705,7 @@ public void testIdempotenceWithOldMagic() throws InterruptedException { long totalSize = 10 * batchSize; String metricGrpName = "producer-metrics"; - apiVersions.update("foobar", NodeApiVersions.create(Arrays.asList(new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE.id, + apiVersions.update("foobar", NodeApiVersions.create(Arrays.asList(new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE, (short) 0, (short) 2)))); RecordAccumulator accum = new RecordAccumulator(logContext, batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, CompressionType.NONE, lingerMs, retryBackoffMs, deliveryTimeoutMs, metrics, metricGrpName, time, apiVersions, new TransactionManager(), diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java index 04197d86ca4c7..5ae76c34a0142 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java @@ -181,7 +181,7 @@ public void testMessageFormatDownConversion() throws Exception { // now the partition leader supports only v2 apiVersions.update("0", NodeApiVersions.create(Collections.singleton( - new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE.id, (short) 0, (short) 2)))); + new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE, (short) 0, (short) 2)))); client.prepareResponse(new MockClient.RequestMatcher() { @Override @@ -221,7 +221,7 @@ public void testDownConversionForMismatchedMagicValues() throws Exception { // now the partition leader supports only v2 apiVersions.update("0", NodeApiVersions.create(Collections.singleton( - new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE.id, (short) 0, (short) 2)))); + new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE, (short) 0, (short) 2)))); Future future2 = accumulator.append(tp1, 0L, "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; diff --git a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java index 547dbc4b9dbbb..7c9d3351a35cc 100644 --- a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java +++ b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java @@ -37,10 +37,10 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.List; import java.util.function.Supplier; +import static java.util.Collections.singletonList; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -67,10 +67,10 @@ public void testAddPartitionsToTxnVersions() throws Exception { setTransactionalId("blah"). setProducerId(0xbadcafebadcafeL). setProducerEpoch((short) 30000). - setTopics(new AddPartitionsToTxnTopicCollection(Collections.singletonList( + setTopics(new AddPartitionsToTxnTopicCollection(singletonList( new AddPartitionsToTxnTopic(). setName("Topic"). - setPartitions(Collections.singletonList(1))).iterator()))); + setPartitions(singletonList(1))).iterator()))); } @Test @@ -97,16 +97,16 @@ public void testMetadataVersions() throws Exception { Arrays.asList(new MetadataRequestData.MetadataRequestTopic().setName("foo"), new MetadataRequestData.MetadataRequestTopic().setName("bar") ))); - testAllMessageRoundTripsFromVersion(new MetadataRequestData(). + testAllMessageRoundTripsFromVersion((short) 1, new MetadataRequestData(). setTopics(null). setAllowAutoTopicCreation(true). setIncludeClusterAuthorizedOperations(false). - setIncludeTopicAuthorizedOperations(false), (short) 1); - testAllMessageRoundTripsFromVersion(new MetadataRequestData(). + setIncludeTopicAuthorizedOperations(false)); + testAllMessageRoundTripsFromVersion((short) 4, new MetadataRequestData(). setTopics(null). setAllowAutoTopicCreation(false). setIncludeClusterAuthorizedOperations(false). - setIncludeTopicAuthorizedOperations(false), (short) 4); + setIncludeTopicAuthorizedOperations(false)); } @Test @@ -117,7 +117,7 @@ public void testHeartbeatVersions() throws Exception { .setGenerationId(15); testAllMessageRoundTrips(newRequest.get()); testAllMessageRoundTrips(newRequest.get().setGroupInstanceId(null)); - testAllMessageRoundTripsFromVersion(newRequest.get().setGroupInstanceId("instanceId"), (short) 3); + testAllMessageRoundTripsFromVersion((short) 3, newRequest.get().setGroupInstanceId("instanceId")); } @Test @@ -129,9 +129,9 @@ public void testJoinGroupVersions() throws Exception { .setProtocols(new JoinGroupRequestData.JoinGroupRequestProtocolCollection()) .setSessionTimeoutMs(10000); testAllMessageRoundTrips(newRequest.get()); - testAllMessageRoundTripsFromVersion(newRequest.get().setRebalanceTimeoutMs(20000), (short) 1); + testAllMessageRoundTripsFromVersion((short) 1, newRequest.get().setRebalanceTimeoutMs(20000)); testAllMessageRoundTrips(newRequest.get().setGroupInstanceId(null)); - testAllMessageRoundTripsFromVersion(newRequest.get().setGroupInstanceId("instanceId"), (short) 5); + testAllMessageRoundTripsFromVersion((short) 5, newRequest.get().setGroupInstanceId("instanceId")); } @Test @@ -143,7 +143,7 @@ public void testSyncGroupDefaultGroupInstanceId() throws Exception { .setAssignments(new ArrayList<>()); testAllMessageRoundTrips(request.get()); testAllMessageRoundTrips(request.get().setGroupInstanceId(null)); - testAllMessageRoundTripsFromVersion(request.get().setGroupInstanceId("instanceId"), (short) 3); + testAllMessageRoundTripsFromVersion((short) 3, request.get().setGroupInstanceId("instanceId")); } @Test @@ -157,27 +157,70 @@ public void testOffsetCommitDefaultGroupInstanceId() throws Exception { .setMemberId("memberId") .setTopics(new ArrayList<>()) .setGenerationId(15); - testAllMessageRoundTripsFromVersion(request.get(), (short) 1); - testAllMessageRoundTripsFromVersion(request.get().setGroupInstanceId(null), (short) 1); - testAllMessageRoundTripsFromVersion(request.get().setGroupInstanceId("instanceId"), (short) 7); + testAllMessageRoundTripsFromVersion((short) 1, request.get()); + testAllMessageRoundTripsFromVersion((short) 1, request.get().setGroupInstanceId(null)); + testAllMessageRoundTripsFromVersion((short) 7, request.get().setGroupInstanceId("instanceId")); + } + + @Test + public void testOffsetForLeaderEpochVersions() throws Exception { + // Version 2 adds optional current leader epoch + OffsetForLeaderEpochRequestData.OffsetForLeaderPartition partitionDataNoCurrentEpoch = + new OffsetForLeaderEpochRequestData.OffsetForLeaderPartition() + .setPartitionIndex(0) + .setLeaderEpoch(3); + OffsetForLeaderEpochRequestData.OffsetForLeaderPartition partitionDataWithCurrentEpoch = + new OffsetForLeaderEpochRequestData.OffsetForLeaderPartition() + .setPartitionIndex(0) + .setLeaderEpoch(3) + .setCurrentLeaderEpoch(5); + + testAllMessageRoundTrips(new OffsetForLeaderEpochRequestData().setTopics(singletonList( + new OffsetForLeaderEpochRequestData.OffsetForLeaderTopic() + .setName("foo") + .setPartitions(singletonList(partitionDataNoCurrentEpoch))) + )); + testAllMessageRoundTripsBeforeVersion((short) 2, partitionDataWithCurrentEpoch, partitionDataNoCurrentEpoch); + testAllMessageRoundTripsFromVersion((short) 2, partitionDataWithCurrentEpoch); + + // Version 3 adds the optional replica Id field + testAllMessageRoundTripsFromVersion((short) 3, new OffsetForLeaderEpochRequestData().setReplicaId(5)); + testAllMessageRoundTripsBeforeVersion((short) 3, + new OffsetForLeaderEpochRequestData().setReplicaId(5), + new OffsetForLeaderEpochRequestData()); + testAllMessageRoundTripsBeforeVersion((short) 3, + new OffsetForLeaderEpochRequestData().setReplicaId(5), + new OffsetForLeaderEpochRequestData().setReplicaId(-2)); + } private void testAllMessageRoundTrips(Message message) throws Exception { - testAllMessageRoundTripsFromVersion(message, message.lowestSupportedVersion()); + testAllMessageRoundTripsFromVersion(message.lowestSupportedVersion(), message); } - private void testAllMessageRoundTripsFromVersion(Message message, short fromVersion) throws Exception { + private void testAllMessageRoundTripsBeforeVersion(short beforeVersion, Message message, Message expected) throws Exception { + for (short version = 0; version < beforeVersion; version++) { + testMessageRoundTrip(version, message, expected); + } + } + + private void testAllMessageRoundTripsFromVersion(short fromVersion, Message message) throws Exception { for (short version = fromVersion; version < message.highestSupportedVersion(); version++) { - testMessageRoundTrips(message, version); + testEquivalentMessageRoundTrip(version, message); } } - private void testMessageRoundTrips(Message message, short version) throws Exception { - testStructRoundTrip(message, version); - testByteBufferRoundTrip(message, version); + private void testMessageRoundTrip(short version, Message message, Message expected) throws Exception { + testByteBufferRoundTrip(version, message, expected); + testStructRoundTrip(version, message, expected); + } + + private void testEquivalentMessageRoundTrip(short version, Message message) throws Exception { + testStructRoundTrip(version, message, message); + testByteBufferRoundTrip(version, message, message); } - private void testByteBufferRoundTrip(Message message, short version) throws Exception { + private void testByteBufferRoundTrip(short version, Message message, Message expected) throws Exception { int size = message.size(version); ByteBuffer buf = ByteBuffer.allocate(size); ByteBufferAccessor byteBufferAccessor = new ByteBufferAccessor(buf); @@ -187,18 +230,18 @@ private void testByteBufferRoundTrip(Message message, short version) throws Exce buf.flip(); message2.read(byteBufferAccessor, version); assertEquals(size, buf.position()); - assertEquals(message, message2); - assertEquals(message.hashCode(), message2.hashCode()); - assertEquals(message.toString(), message2.toString()); + assertEquals(expected, message2); + assertEquals(expected.hashCode(), message2.hashCode()); + assertEquals(expected.toString(), message2.toString()); } - private void testStructRoundTrip(Message message, short version) throws Exception { + private void testStructRoundTrip(short version, Message message, Message expected) throws Exception { Struct struct = message.toStruct(version); Message message2 = message.getClass().newInstance(); message2.fromStruct(struct, version); - assertEquals(message, message2); - assertEquals(message.hashCode(), message2.hashCode()); - assertEquals(message.toString(), message2.toString()); + assertEquals(expected, message2); + assertEquals(expected.hashCode(), message2.hashCode()); + assertEquals(expected.toString(), message2.toString()); } /** @@ -348,7 +391,7 @@ private static void compareTypes(NamedType typeA, NamedType typeB) { */ private static List flatten(NamedType type) { if (!(type.type instanceof Schema)) { - return Collections.singletonList(type); + return singletonList(type); } Schema schema = (Schema) type.type; ArrayList results = new ArrayList<>(); @@ -367,7 +410,7 @@ public void testDefaultValues() throws Exception { verifySizeSucceeds((short) 0, new OffsetCommitRequestData().setRetentionTimeMs(123)); verifySizeRaisesUve((short) 5, "forgotten", - new FetchRequestData().setForgotten(Collections.singletonList( + new FetchRequestData().setForgotten(singletonList( new FetchRequestData.ForgottenTopic().setName("foo")))); } diff --git a/clients/src/test/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequestTest.java new file mode 100644 index 0000000000000..604f88f04aeaa --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequestTest.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.requests; + +import org.apache.kafka.common.errors.UnsupportedVersionException; +import org.apache.kafka.common.protocol.ApiKeys; +import org.junit.Test; + +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +public class OffsetsForLeaderEpochRequestTest { + + @Test + public void testForConsumerRequiresVersion3() { + OffsetsForLeaderEpochRequest.Builder builder = OffsetsForLeaderEpochRequest.Builder.forConsumer(Collections.emptyMap()); + for (short version = 0; version < 3; version++) { + final short v = version; + assertThrows(UnsupportedVersionException.class, () -> builder.build(v)); + } + + for (short version = 3; version < ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(); version++) { + OffsetsForLeaderEpochRequest request = builder.build((short) 3); + assertEquals(OffsetsForLeaderEpochRequest.CONSUMER_REPLICA_ID, request.replicaId()); + } + } + + @Test + public void testDefaultReplicaId() { + for (short version = 0; version < ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(); version++) { + int replicaId = 1; + OffsetsForLeaderEpochRequest.Builder builder = OffsetsForLeaderEpochRequest.Builder.forFollower( + version, Collections.emptyMap(), replicaId); + OffsetsForLeaderEpochRequest request = builder.build(); + OffsetsForLeaderEpochRequest parsed = (OffsetsForLeaderEpochRequest) AbstractRequest.parseRequest( + ApiKeys.OFFSET_FOR_LEADER_EPOCH, version, request.toStruct()); + if (version < 3) + assertEquals(OffsetsForLeaderEpochRequest.DEBUGGING_REPLICA_ID, parsed.replicaId()); + else + assertEquals(replicaId, parsed.replicaId()); + } + } + +} \ No newline at end of file diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index e8f349f594766..32f3305b33598 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -290,10 +290,10 @@ public void testSerialization() throws Exception { checkRequest(createListOffsetRequest(0), true); checkErrorResponse(createListOffsetRequest(0), new UnknownServerException(), true); checkResponse(createListOffsetResponse(0), 0, true); - checkRequest(createLeaderEpochRequest(0), true); - checkRequest(createLeaderEpochRequest(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion()), true); + checkRequest(createLeaderEpochRequestForReplica(0, 1), true); + checkRequest(createLeaderEpochRequestForConsumer(), true); checkResponse(createLeaderEpochResponse(), 0, true); - checkErrorResponse(createLeaderEpochRequest(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion()), new UnknownServerException(), true); + checkErrorResponse(createLeaderEpochRequestForConsumer(), new UnknownServerException(), true); checkRequest(createAddPartitionsToTxnRequest(), true); checkErrorResponse(createAddPartitionsToTxnRequest(), new UnknownServerException(), true); checkResponse(createAddPartitionsToTxnResponse(), 0, true); @@ -1251,17 +1251,25 @@ private InitProducerIdResponse createInitPidResponse() { return new InitProducerIdResponse(responseData); } - private OffsetsForLeaderEpochRequest createLeaderEpochRequest(int version) { + private Map createOffsetForLeaderEpochPartitionData() { Map epochs = new HashMap<>(); - epochs.put(new TopicPartition("topic1", 0), new OffsetsForLeaderEpochRequest.PartitionData(Optional.of(0), 1)); epochs.put(new TopicPartition("topic1", 1), new OffsetsForLeaderEpochRequest.PartitionData(Optional.of(0), 1)); epochs.put(new TopicPartition("topic2", 2), new OffsetsForLeaderEpochRequest.PartitionData(Optional.empty(), 3)); + return epochs; + } + + private OffsetsForLeaderEpochRequest createLeaderEpochRequestForConsumer() { + Map epochs = createOffsetForLeaderEpochPartitionData(); + return OffsetsForLeaderEpochRequest.Builder.forConsumer(epochs).build(); + } - return OffsetsForLeaderEpochRequest.Builder.forConsumer((short) version, epochs).build(); + private OffsetsForLeaderEpochRequest createLeaderEpochRequestForReplica(int version, int replicaId) { + Map epochs = createOffsetForLeaderEpochPartitionData(); + return OffsetsForLeaderEpochRequest.Builder.forFollower((short) version, epochs, replicaId).build(); } private OffsetsForLeaderEpochResponse createLeaderEpochResponse() { diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index a577b2e99717f..011a19b27a6c8 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -295,7 +295,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest { private def offsetsForLeaderEpochRequest: OffsetsForLeaderEpochRequest = { val epochs = Map(tp -> new OffsetsForLeaderEpochRequest.PartitionData(Optional.of(27), 7)) - OffsetsForLeaderEpochRequest.Builder.forConsumer(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, epochs.asJava).build() + OffsetsForLeaderEpochRequest.Builder.forConsumer(epochs.asJava).build() } private def createOffsetFetchRequest = { diff --git a/core/src/test/scala/unit/kafka/server/OffsetsForLeaderEpochRequestTest.scala b/core/src/test/scala/unit/kafka/server/OffsetsForLeaderEpochRequestTest.scala index 4d1416cdd0298..b6f8a142012ab 100644 --- a/core/src/test/scala/unit/kafka/server/OffsetsForLeaderEpochRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/OffsetsForLeaderEpochRequestTest.scala @@ -35,7 +35,8 @@ class OffsetsForLeaderEpochRequestTest extends BaseRequestTest { val partition = new TopicPartition(topic, 0) val epochs = Map(partition -> new OffsetsForLeaderEpochRequest.PartitionData(Optional.empty[Integer], 0)).asJava - val request = OffsetsForLeaderEpochRequest.Builder.forConsumer(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, epochs).build() + val request = OffsetsForLeaderEpochRequest.Builder.forFollower( + ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, epochs, 1).build() // Unknown topic val randomBrokerId = servers.head.config.brokerId @@ -61,8 +62,8 @@ class OffsetsForLeaderEpochRequestTest extends BaseRequestTest { def assertResponseErrorForEpoch(error: Errors, brokerId: Int, currentLeaderEpoch: Optional[Integer]): Unit = { val epochs = Map(topicPartition -> new OffsetsForLeaderEpochRequest.PartitionData( currentLeaderEpoch, 0)).asJava - val request = OffsetsForLeaderEpochRequest.Builder.forConsumer(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, epochs) - .build() + val request = OffsetsForLeaderEpochRequest.Builder.forFollower( + ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, epochs, 1).build() assertResponseError(error, brokerId, request) } diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala index 047188f304f88..a7fbc5d198ca9 100644 --- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala +++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala @@ -362,8 +362,8 @@ class RequestQuotaTest extends BaseRequestTest { new InitProducerIdRequest.Builder(requestData) case ApiKeys.OFFSET_FOR_LEADER_EPOCH => - OffsetsForLeaderEpochRequest.Builder.forConsumer(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, - Map(tp -> new OffsetsForLeaderEpochRequest.PartitionData(Optional.of(15), 0)).asJava) + OffsetsForLeaderEpochRequest.Builder.forConsumer(Map(tp -> + new OffsetsForLeaderEpochRequest.PartitionData(Optional.of(15), 0)).asJava) case ApiKeys.ADD_PARTITIONS_TO_TXN => new AddPartitionsToTxnRequest.Builder("test-transactional-id", 1, 0, List(tp).asJava) diff --git a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala index 8a6dcba7d9209..cf78bac8d617c 100644 --- a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala @@ -279,8 +279,9 @@ class LeaderEpochIntegrationTest extends ZooKeeperTestHarness with Logging { def leaderOffsetsFor(partitions: Map[TopicPartition, Int]): Map[TopicPartition, EpochEndOffset] = { val partitionData = partitions.mapValues( new OffsetsForLeaderEpochRequest.PartitionData(Optional.empty(), _)) - val request = OffsetsForLeaderEpochRequest.Builder.forConsumer(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, - partitionData.asJava) + + val request = OffsetsForLeaderEpochRequest.Builder.forFollower( + ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion, partitionData.asJava, 1) val response = sender.sendRequest(request) response.responseBody.asInstanceOf[OffsetsForLeaderEpochResponse].responses.asScala }