chia7712 commented on code in PR #22252: URL: https://github.com/apache/kafka/pull/22252#discussion_r4477543363
########## server/src/test/java/org/apache/kafka/server/requests/FetchFromFollowerIntegrationTest.java: ########## @@ -0,0 +1,461 @@ +/* + * 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.server.requests; + +import kafka.server.KafkaBroker; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewPartitionReassignment; +import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.GroupProtocol; +import org.apache.kafka.clients.consumer.RangeAssignor; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.FetchResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.FetchRequest; +import org.apache.kafka.common.requests.FetchResponse; +import org.apache.kafka.common.test.ClusterInstance; +import org.apache.kafka.common.test.api.ClusterConfigProperty; +import org.apache.kafka.common.test.api.ClusterTest; +import org.apache.kafka.common.test.api.ClusterTestDefaults; +import org.apache.kafka.common.test.api.Type; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.server.IntegrationTestUtils; +import org.apache.kafka.server.TestUtils; + +import org.junit.jupiter.api.Timeout; + +import java.net.Socket; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG; +import static org.apache.kafka.server.TestUtils.consumeRecords; +import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.BROKER_RACK_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG; +import static org.apache.kafka.server.config.ServerLogConfigs.NUM_PARTITIONS_CONFIG; +import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS; +import static org.apache.kafka.test.TestUtils.waitForCondition; +import static org.junit.jupiter.api.Assertions.assertEquals; + +@ClusterTestDefaults( + types = {Type.KRAFT}, + brokers = FetchFromFollowerIntegrationTest.BROKER_COUNT, + serverProperties = { + @ClusterConfigProperty(id = 0, key = BROKER_RACK_CONFIG, value = "0"), + @ClusterConfigProperty(id = 1, key = BROKER_RACK_CONFIG, value = "1"), + @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, + value = "org.apache.kafka.common.replica.RackAwareReplicaSelector"), + @ClusterConfigProperty(key = CONTROLLED_SHUTDOWN_ENABLE_CONFIG, value = "false"), + @ClusterConfigProperty(key = NUM_PARTITIONS_CONFIG, value = "1"), + @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"), + } +) +public class FetchFromFollowerIntegrationTest { + + public static final int BROKER_COUNT = 2; + public static final String TOPIC = "test-fetch-from-follower"; + public static final int LEADER_BROKER_ID = 0; + public static final int FOLLOWER_BROKER_ID = 1; + + @ClusterTest + @Timeout(15) + public void testFollowerCompleteDelayedFetchesOnReplication(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + int leaderId = waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + var topicPartition = new TopicPartition(TOPIC, 0); + assertEquals(LEADER_BROKER_ID, leaderId); + + // Set fetch.max.wait.ms to a value (20 seconds) greater than the timeout (15 seconds). + // Send a fetch request before the record is replicated to ensure that the replication + // triggers purgatory completion. + LinkedHashMap<TopicPartition, FetchRequest.PartitionData> fetchData = createPartitionMap(cluster, 1000, List.of(topicPartition), Map.of(topicPartition, 0L)); + FetchRequest fetchRequest = FetchRequest.Builder.forConsumer(ApiKeys.FETCH.latestVersion(), 20000, 1, fetchData) + .setMaxBytes(1000) + .rackId("") + .build(); + + int leaderPort = cluster.brokers().get(LEADER_BROKER_ID).boundPort(cluster.clientListener()); + int followerPort = cluster.brokers().get(FOLLOWER_BROKER_ID).boundPort(cluster.clientListener()); + byte[] serializedFetchRequest = Utils.toArray(fetchRequest.serializeWithHeader( + IntegrationTestUtils.nextRequestHeader(ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()))); + try (Socket leaderSocket = IntegrationTestUtils.connect(leaderPort); + Socket followerSocket = IntegrationTestUtils.connect(followerPort)) { + IntegrationTestUtils.sendRequest(leaderSocket, serializedFetchRequest); + IntegrationTestUtils.sendRequest(followerSocket, serializedFetchRequest); + + try (Producer<byte[], byte[]> producer = cluster.producer(Map.of())) { + producer.send(new ProducerRecord<>(TOPIC, "key".getBytes(), "value".getBytes())).get(); + } + + FetchResponse leaderResponse = IntegrationTestUtils.receive(leaderSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, leaderResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), leaderResponse.errorCounts()); + + FetchResponse followerResponse = IntegrationTestUtils.receive(followerSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, followerResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), followerResponse.errorCounts()); + } + } + + @ClusterTest + public void testFetchFromLeaderWhilePreferredReadReplicaIsUnavailable(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + produceMessages(cluster, TOPIC, 10); + assertEquals(1, getPreferredReplica(cluster, TOPIC)); + + cluster.brokers().get(FOLLOWER_BROKER_ID).shutdown(); Review Comment: Could you use `cluster.shutdownBroker();` instead ########## server/src/test/java/org/apache/kafka/server/TestUtils.java: ########## @@ -181,4 +181,38 @@ public static int awaitLeaderChange(ClusterInstance cluster, TopicPartition tp, return newLeaderExists.get().get(); } + + /** + * Wait for the leader of the partition to change from {@code oldLeaderOpt} to {@code expectedLeaderOpt}, + * or for any leader to be elected if both are empty. + * + * @param brokers The list of brokers + * @param tp The topic partition to check + * @param oldLeaderOpt If defined, wait for the leader to change away from this broker + * @param expectedLeaderOpt If defined, wait for this specific broker to become leader + * @param timeoutMs The duration in ms to wait + * @return The elected leader broker id + */ + public static int awaitLeaderChange( Review Comment: It seems the usage could be replaced by `TestUtils.awaitLeaderChange(cluster, tp, Optional.empty())` ########## clients/src/test/java/org/apache/kafka/common/requests/FetchRequestTest.java: ########## @@ -28,7 +28,6 @@ import org.junit.jupiter.params.provider.Arguments; Review Comment: Please revert those unrelated changes :) ########## server/src/test/java/org/apache/kafka/server/requests/FetchFromFollowerIntegrationTest.java: ########## @@ -0,0 +1,461 @@ +/* + * 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.server.requests; + +import kafka.server.KafkaBroker; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewPartitionReassignment; +import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.GroupProtocol; +import org.apache.kafka.clients.consumer.RangeAssignor; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.FetchResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.FetchRequest; +import org.apache.kafka.common.requests.FetchResponse; +import org.apache.kafka.common.test.ClusterInstance; +import org.apache.kafka.common.test.api.ClusterConfigProperty; +import org.apache.kafka.common.test.api.ClusterTest; +import org.apache.kafka.common.test.api.ClusterTestDefaults; +import org.apache.kafka.common.test.api.Type; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.server.IntegrationTestUtils; +import org.apache.kafka.server.TestUtils; + +import org.junit.jupiter.api.Timeout; + +import java.net.Socket; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG; +import static org.apache.kafka.server.TestUtils.consumeRecords; +import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.BROKER_RACK_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG; +import static org.apache.kafka.server.config.ServerLogConfigs.NUM_PARTITIONS_CONFIG; +import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS; +import static org.apache.kafka.test.TestUtils.waitForCondition; +import static org.junit.jupiter.api.Assertions.assertEquals; + +@ClusterTestDefaults( + types = {Type.KRAFT}, + brokers = FetchFromFollowerIntegrationTest.BROKER_COUNT, + serverProperties = { + @ClusterConfigProperty(id = 0, key = BROKER_RACK_CONFIG, value = "0"), + @ClusterConfigProperty(id = 1, key = BROKER_RACK_CONFIG, value = "1"), + @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, + value = "org.apache.kafka.common.replica.RackAwareReplicaSelector"), + @ClusterConfigProperty(key = CONTROLLED_SHUTDOWN_ENABLE_CONFIG, value = "false"), + @ClusterConfigProperty(key = NUM_PARTITIONS_CONFIG, value = "1"), + @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"), + } +) +public class FetchFromFollowerIntegrationTest { + + public static final int BROKER_COUNT = 2; + public static final String TOPIC = "test-fetch-from-follower"; + public static final int LEADER_BROKER_ID = 0; + public static final int FOLLOWER_BROKER_ID = 1; + + @ClusterTest + @Timeout(15) + public void testFollowerCompleteDelayedFetchesOnReplication(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + int leaderId = waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + var topicPartition = new TopicPartition(TOPIC, 0); + assertEquals(LEADER_BROKER_ID, leaderId); + + // Set fetch.max.wait.ms to a value (20 seconds) greater than the timeout (15 seconds). + // Send a fetch request before the record is replicated to ensure that the replication + // triggers purgatory completion. + LinkedHashMap<TopicPartition, FetchRequest.PartitionData> fetchData = createPartitionMap(cluster, 1000, List.of(topicPartition), Map.of(topicPartition, 0L)); + FetchRequest fetchRequest = FetchRequest.Builder.forConsumer(ApiKeys.FETCH.latestVersion(), 20000, 1, fetchData) + .setMaxBytes(1000) + .rackId("") + .build(); + + int leaderPort = cluster.brokers().get(LEADER_BROKER_ID).boundPort(cluster.clientListener()); + int followerPort = cluster.brokers().get(FOLLOWER_BROKER_ID).boundPort(cluster.clientListener()); + byte[] serializedFetchRequest = Utils.toArray(fetchRequest.serializeWithHeader( + IntegrationTestUtils.nextRequestHeader(ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()))); + try (Socket leaderSocket = IntegrationTestUtils.connect(leaderPort); + Socket followerSocket = IntegrationTestUtils.connect(followerPort)) { + IntegrationTestUtils.sendRequest(leaderSocket, serializedFetchRequest); + IntegrationTestUtils.sendRequest(followerSocket, serializedFetchRequest); + + try (Producer<byte[], byte[]> producer = cluster.producer(Map.of())) { + producer.send(new ProducerRecord<>(TOPIC, "key".getBytes(), "value".getBytes())).get(); + } + + FetchResponse leaderResponse = IntegrationTestUtils.receive(leaderSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, leaderResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), leaderResponse.errorCounts()); + + FetchResponse followerResponse = IntegrationTestUtils.receive(followerSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, followerResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), followerResponse.errorCounts()); + } + } + + @ClusterTest + public void testFetchFromLeaderWhilePreferredReadReplicaIsUnavailable(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + produceMessages(cluster, TOPIC, 10); + assertEquals(1, getPreferredReplica(cluster, TOPIC)); + + cluster.brokers().get(FOLLOWER_BROKER_ID).shutdown(); + TopicPartition topicPartition = new TopicPartition(TOPIC, 0); + + waitForCondition( + () -> { + KafkaBroker leaderBroker = cluster.brokers().get(LEADER_BROKER_ID); + Map<Integer, Node> endpoints = leaderBroker.metadataCache() + .getPartitionReplicaEndpoints(topicPartition, cluster.clientListener()); + return !endpoints.containsKey(FOLLOWER_BROKER_ID); + }, + "follower is still reachable." + ); + + assertEquals(-1, getPreferredReplica(cluster, TOPIC)); + } + + @ClusterTest + public void testFetchFromFollowerWithRoll(ClusterInstance cluster) throws Exception { + for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) { Review Comment: Could you split it into two test cases? Otherwise, the latter one won’t run if the former fails. ########## server/src/test/java/org/apache/kafka/server/requests/FetchFromFollowerIntegrationTest.java: ########## @@ -0,0 +1,461 @@ +/* + * 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.server.requests; + +import kafka.server.KafkaBroker; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewPartitionReassignment; +import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.GroupProtocol; +import org.apache.kafka.clients.consumer.RangeAssignor; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.FetchResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.FetchRequest; +import org.apache.kafka.common.requests.FetchResponse; +import org.apache.kafka.common.test.ClusterInstance; +import org.apache.kafka.common.test.api.ClusterConfigProperty; +import org.apache.kafka.common.test.api.ClusterTest; +import org.apache.kafka.common.test.api.ClusterTestDefaults; +import org.apache.kafka.common.test.api.Type; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.server.IntegrationTestUtils; +import org.apache.kafka.server.TestUtils; + +import org.junit.jupiter.api.Timeout; + +import java.net.Socket; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG; +import static org.apache.kafka.server.TestUtils.consumeRecords; +import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.BROKER_RACK_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG; +import static org.apache.kafka.server.config.ServerLogConfigs.NUM_PARTITIONS_CONFIG; +import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS; +import static org.apache.kafka.test.TestUtils.waitForCondition; +import static org.junit.jupiter.api.Assertions.assertEquals; + +@ClusterTestDefaults( + types = {Type.KRAFT}, + brokers = FetchFromFollowerIntegrationTest.BROKER_COUNT, + serverProperties = { + @ClusterConfigProperty(id = 0, key = BROKER_RACK_CONFIG, value = "0"), + @ClusterConfigProperty(id = 1, key = BROKER_RACK_CONFIG, value = "1"), + @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, + value = "org.apache.kafka.common.replica.RackAwareReplicaSelector"), + @ClusterConfigProperty(key = CONTROLLED_SHUTDOWN_ENABLE_CONFIG, value = "false"), + @ClusterConfigProperty(key = NUM_PARTITIONS_CONFIG, value = "1"), + @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"), + } +) +public class FetchFromFollowerIntegrationTest { + + public static final int BROKER_COUNT = 2; + public static final String TOPIC = "test-fetch-from-follower"; + public static final int LEADER_BROKER_ID = 0; + public static final int FOLLOWER_BROKER_ID = 1; + + @ClusterTest + @Timeout(15) + public void testFollowerCompleteDelayedFetchesOnReplication(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + int leaderId = waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + var topicPartition = new TopicPartition(TOPIC, 0); + assertEquals(LEADER_BROKER_ID, leaderId); + + // Set fetch.max.wait.ms to a value (20 seconds) greater than the timeout (15 seconds). + // Send a fetch request before the record is replicated to ensure that the replication + // triggers purgatory completion. + LinkedHashMap<TopicPartition, FetchRequest.PartitionData> fetchData = createPartitionMap(cluster, 1000, List.of(topicPartition), Map.of(topicPartition, 0L)); + FetchRequest fetchRequest = FetchRequest.Builder.forConsumer(ApiKeys.FETCH.latestVersion(), 20000, 1, fetchData) + .setMaxBytes(1000) + .rackId("") + .build(); + + int leaderPort = cluster.brokers().get(LEADER_BROKER_ID).boundPort(cluster.clientListener()); + int followerPort = cluster.brokers().get(FOLLOWER_BROKER_ID).boundPort(cluster.clientListener()); + byte[] serializedFetchRequest = Utils.toArray(fetchRequest.serializeWithHeader( + IntegrationTestUtils.nextRequestHeader(ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()))); + try (Socket leaderSocket = IntegrationTestUtils.connect(leaderPort); + Socket followerSocket = IntegrationTestUtils.connect(followerPort)) { + IntegrationTestUtils.sendRequest(leaderSocket, serializedFetchRequest); + IntegrationTestUtils.sendRequest(followerSocket, serializedFetchRequest); + + try (Producer<byte[], byte[]> producer = cluster.producer(Map.of())) { + producer.send(new ProducerRecord<>(TOPIC, "key".getBytes(), "value".getBytes())).get(); + } + + FetchResponse leaderResponse = IntegrationTestUtils.receive(leaderSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, leaderResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), leaderResponse.errorCounts()); + + FetchResponse followerResponse = IntegrationTestUtils.receive(followerSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, followerResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), followerResponse.errorCounts()); + } + } + + @ClusterTest + public void testFetchFromLeaderWhilePreferredReadReplicaIsUnavailable(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + produceMessages(cluster, TOPIC, 10); + assertEquals(1, getPreferredReplica(cluster, TOPIC)); + + cluster.brokers().get(FOLLOWER_BROKER_ID).shutdown(); + TopicPartition topicPartition = new TopicPartition(TOPIC, 0); + + waitForCondition( + () -> { + KafkaBroker leaderBroker = cluster.brokers().get(LEADER_BROKER_ID); + Map<Integer, Node> endpoints = leaderBroker.metadataCache() + .getPartitionReplicaEndpoints(topicPartition, cluster.clientListener()); + return !endpoints.containsKey(FOLLOWER_BROKER_ID); + }, + "follower is still reachable." + ); + + assertEquals(-1, getPreferredReplica(cluster, TOPIC)); + } + + @ClusterTest + public void testFetchFromFollowerWithRoll(ClusterInstance cluster) throws Exception { + for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) { + verifyFetchFromFollowerWithRoll(cluster, groupProtocol); + } + } + + private void verifyFetchFromFollowerWithRoll(ClusterInstance cluster, GroupProtocol groupProtocol) throws Exception { + // Each protocol gets its own topic and group ids so that the iterations of + // testFetchFromFollowerWithRoll, which share a single cluster, stay independent. + String protocolName = groupProtocol.name().toLowerCase(Locale.ROOT); + String topic = TOPIC + "-" + protocolName; + cluster.createTopicWithAssignment(topic, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(topic, 0)); + + var followerConsumerProps = new HashMap<String, Object>(); + followerConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()); + followerConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-follower-" + protocolName); + followerConsumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + followerConsumerProps.put(ConsumerConfig.CLIENT_RACK_CONFIG, String.valueOf(FOLLOWER_BROKER_ID)); + followerConsumerProps.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name()); + + var leaderConsumerProps = new HashMap<String, Object>(); + leaderConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()); + leaderConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-leader-" + protocolName); + leaderConsumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + leaderConsumerProps.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name()); + + try (Consumer<byte[], byte[]> followerConsumer = cluster.consumer(followerConsumerProps); + Consumer<byte[], byte[]> leaderConsumer = cluster.consumer(leaderConsumerProps)) { Review Comment: This consumer has no rack, so it seems it is unrelated to this test case. Could you remove it? ########## server/src/test/java/org/apache/kafka/server/requests/FetchFromFollowerIntegrationTest.java: ########## @@ -0,0 +1,461 @@ +/* + * 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.server.requests; + +import kafka.server.KafkaBroker; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewPartitionReassignment; +import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.GroupProtocol; +import org.apache.kafka.clients.consumer.RangeAssignor; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.FetchResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.FetchRequest; +import org.apache.kafka.common.requests.FetchResponse; +import org.apache.kafka.common.test.ClusterInstance; +import org.apache.kafka.common.test.api.ClusterConfigProperty; +import org.apache.kafka.common.test.api.ClusterTest; +import org.apache.kafka.common.test.api.ClusterTestDefaults; +import org.apache.kafka.common.test.api.Type; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.server.IntegrationTestUtils; +import org.apache.kafka.server.TestUtils; + +import org.junit.jupiter.api.Timeout; + +import java.net.Socket; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG; +import static org.apache.kafka.server.TestUtils.consumeRecords; +import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.BROKER_RACK_CONFIG; +import static org.apache.kafka.server.config.ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG; +import static org.apache.kafka.server.config.ServerLogConfigs.NUM_PARTITIONS_CONFIG; +import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS; +import static org.apache.kafka.test.TestUtils.waitForCondition; +import static org.junit.jupiter.api.Assertions.assertEquals; + +@ClusterTestDefaults( + types = {Type.KRAFT}, + brokers = FetchFromFollowerIntegrationTest.BROKER_COUNT, + serverProperties = { + @ClusterConfigProperty(id = 0, key = BROKER_RACK_CONFIG, value = "0"), + @ClusterConfigProperty(id = 1, key = BROKER_RACK_CONFIG, value = "1"), + @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, + value = "org.apache.kafka.common.replica.RackAwareReplicaSelector"), + @ClusterConfigProperty(key = CONTROLLED_SHUTDOWN_ENABLE_CONFIG, value = "false"), + @ClusterConfigProperty(key = NUM_PARTITIONS_CONFIG, value = "1"), + @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"), + } +) +public class FetchFromFollowerIntegrationTest { + + public static final int BROKER_COUNT = 2; + public static final String TOPIC = "test-fetch-from-follower"; + public static final int LEADER_BROKER_ID = 0; + public static final int FOLLOWER_BROKER_ID = 1; + + @ClusterTest + @Timeout(15) + public void testFollowerCompleteDelayedFetchesOnReplication(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + int leaderId = waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + var topicPartition = new TopicPartition(TOPIC, 0); + assertEquals(LEADER_BROKER_ID, leaderId); + + // Set fetch.max.wait.ms to a value (20 seconds) greater than the timeout (15 seconds). + // Send a fetch request before the record is replicated to ensure that the replication + // triggers purgatory completion. + LinkedHashMap<TopicPartition, FetchRequest.PartitionData> fetchData = createPartitionMap(cluster, 1000, List.of(topicPartition), Map.of(topicPartition, 0L)); + FetchRequest fetchRequest = FetchRequest.Builder.forConsumer(ApiKeys.FETCH.latestVersion(), 20000, 1, fetchData) + .setMaxBytes(1000) + .rackId("") + .build(); + + int leaderPort = cluster.brokers().get(LEADER_BROKER_ID).boundPort(cluster.clientListener()); + int followerPort = cluster.brokers().get(FOLLOWER_BROKER_ID).boundPort(cluster.clientListener()); + byte[] serializedFetchRequest = Utils.toArray(fetchRequest.serializeWithHeader( + IntegrationTestUtils.nextRequestHeader(ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()))); + try (Socket leaderSocket = IntegrationTestUtils.connect(leaderPort); + Socket followerSocket = IntegrationTestUtils.connect(followerPort)) { + IntegrationTestUtils.sendRequest(leaderSocket, serializedFetchRequest); + IntegrationTestUtils.sendRequest(followerSocket, serializedFetchRequest); + + try (Producer<byte[], byte[]> producer = cluster.producer(Map.of())) { + producer.send(new ProducerRecord<>(TOPIC, "key".getBytes(), "value".getBytes())).get(); + } + + FetchResponse leaderResponse = IntegrationTestUtils.receive(leaderSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, leaderResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), leaderResponse.errorCounts()); + + FetchResponse followerResponse = IntegrationTestUtils.receive(followerSocket, ApiKeys.FETCH, ApiKeys.FETCH.latestVersion()); + assertEquals(Errors.NONE, followerResponse.error()); + assertEquals(Map.of(Errors.NONE, 2), followerResponse.errorCounts()); + } + } + + @ClusterTest + public void testFetchFromLeaderWhilePreferredReadReplicaIsUnavailable(ClusterInstance cluster) throws Exception { + cluster.createTopicWithAssignment(TOPIC, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(TOPIC, 0)); + + produceMessages(cluster, TOPIC, 10); + assertEquals(1, getPreferredReplica(cluster, TOPIC)); + + cluster.brokers().get(FOLLOWER_BROKER_ID).shutdown(); + TopicPartition topicPartition = new TopicPartition(TOPIC, 0); + + waitForCondition( + () -> { + KafkaBroker leaderBroker = cluster.brokers().get(LEADER_BROKER_ID); + Map<Integer, Node> endpoints = leaderBroker.metadataCache() + .getPartitionReplicaEndpoints(topicPartition, cluster.clientListener()); + return !endpoints.containsKey(FOLLOWER_BROKER_ID); + }, + "follower is still reachable." + ); + + assertEquals(-1, getPreferredReplica(cluster, TOPIC)); + } + + @ClusterTest + public void testFetchFromFollowerWithRoll(ClusterInstance cluster) throws Exception { + for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) { + verifyFetchFromFollowerWithRoll(cluster, groupProtocol); + } + } + + private void verifyFetchFromFollowerWithRoll(ClusterInstance cluster, GroupProtocol groupProtocol) throws Exception { + // Each protocol gets its own topic and group ids so that the iterations of + // testFetchFromFollowerWithRoll, which share a single cluster, stay independent. + String protocolName = groupProtocol.name().toLowerCase(Locale.ROOT); + String topic = TOPIC + "-" + protocolName; + cluster.createTopicWithAssignment(topic, Map.of(0, List.of(LEADER_BROKER_ID, FOLLOWER_BROKER_ID))); + waitUntilLeaderIsKnown(cluster.brokers().values(), new TopicPartition(topic, 0)); + + var followerConsumerProps = new HashMap<String, Object>(); + followerConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()); + followerConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-follower-" + protocolName); + followerConsumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + followerConsumerProps.put(ConsumerConfig.CLIENT_RACK_CONFIG, String.valueOf(FOLLOWER_BROKER_ID)); + followerConsumerProps.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name()); + + var leaderConsumerProps = new HashMap<String, Object>(); + leaderConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()); + leaderConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-leader-" + protocolName); + leaderConsumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + leaderConsumerProps.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name()); + + try (Consumer<byte[], byte[]> followerConsumer = cluster.consumer(followerConsumerProps); + Consumer<byte[], byte[]> leaderConsumer = cluster.consumer(leaderConsumerProps)) { + followerConsumer.subscribe(List.of(topic)); + leaderConsumer.subscribe(List.of(topic)); + + // Wait until preferred replica is set to follower. + waitForCondition( + () -> getPreferredReplica(cluster, topic) == FOLLOWER_BROKER_ID, + "Preferred replica is not set to follower for " + protocolName + ); + + // Produce and consume from both consumers. + produceMessages(cluster, topic, 1); + consumeRecords(followerConsumer, 1); + consumeRecords(leaderConsumer, 1); + + // Shutdown follower, produce and consume should work for both consumers. + cluster.shutdownBroker(FOLLOWER_BROKER_ID); + produceMessages(cluster, topic, 1); + consumeRecords(followerConsumer, 1); + consumeRecords(leaderConsumer, 1); + + // Start the follower and wait until preferred replica is set to follower. + cluster.startBroker(FOLLOWER_BROKER_ID); + waitForCondition( + () -> getPreferredReplica(cluster, topic) == FOLLOWER_BROKER_ID, + "Preferred replica is not set to follower after restart for " + protocolName + ); + + // Produce and consume should still work for both consumers. + produceMessages(cluster, topic, 1); + consumeRecords(followerConsumer, 1); + consumeRecords(leaderConsumer, 1); + } + } + + @ClusterTest + public void testRackAwareRangeAssignor(ClusterInstance cluster) throws Exception { + List<Integer> partitionList = cluster.brokers().keySet().stream() + .sorted() + .toList(); + + String topicWithAllPartitionsOnAllRacks = "topicWithAllPartitionsOnAllRacks"; + cluster.createTopic(topicWithAllPartitionsOnAllRacks, + cluster.brokers().size(), + (short) cluster.brokers().size()); + + // Racks are in order of broker ids, assign leaders in reverse order + String topicWithSingleRackPartitions = "topicWithSingleRackPartitions"; + var replicaAssignment = new LinkedHashMap<Integer, List<Integer>>(); + for (int i : partitionList) { + replicaAssignment.put(i, List.of(cluster.brokers().size() - i - 1)); + } + cluster.createTopicWithAssignment(topicWithSingleRackPartitions, replicaAssignment); + + List<HashMap<String, Object>> consumerConfigs = cluster.brokers().keySet().stream() + .sorted() + .map(brokerId -> { + var config = new HashMap<String, Object>(); + config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()); + config.put(ConsumerConfig.GROUP_ID_CONFIG, "rack-aware-group"); + config.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName()); + config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + config.put(ConsumerConfig.CLIENT_RACK_CONFIG, String.valueOf(cluster.brokers().get(brokerId).config().rack().orElse(null))); + config.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "instance-" + brokerId); + config.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000"); + config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); + return config; + }) + .toList(); + + List<Consumer<byte[], byte[]>> consumers = consumerConfigs.stream() + .map(cluster::<byte[], byte[]>consumer) + .toList(); + + try (Producer<byte[], byte[]> producer = cluster.producer(Map.of())) { + ExecutorService executor = Executors.newFixedThreadPool(consumers.size()); + try { + // Rack-based assignment results in partitions assigned in reverse order since partition racks are in the reverse order. + verifyRackAwareAssignments(executor, consumers, producer, partitionList, topicWithSingleRackPartitions, reverseList(partitionList), List.of()); + // Non-rack-aware assignment results in ordered partitions. + verifyRackAwareAssignments(executor, consumers, producer, partitionList, topicWithAllPartitionsOnAllRacks, partitionList, List.of()); + // Rack-aware assignment with co-partitioning results in reverse assignment for both topics. + verifyRackAwareAssignments(executor, consumers, producer, partitionList, topicWithAllPartitionsOnAllRacks, reverseList(partitionList), List.of(topicWithSingleRackPartitions)); + + // Perform reassignment for topicWithSingleRackPartitions to reverse the replica racks and + // verify that change in replica racks results in re-assignment based on new racks. + try (Admin admin = cluster.admin()) { + var reassignments = new HashMap<TopicPartition, Optional<NewPartitionReassignment>>(); + for (int p : partitionList) { + var newAssignment = new NewPartitionReassignment(List.of(p)); + reassignments.put(new TopicPartition(topicWithSingleRackPartitions, p), Optional.of(newAssignment)); + } + admin.alterPartitionReassignments(reassignments).all().get(30, TimeUnit.SECONDS); + } + + verifyRackAwareAssignments(executor, consumers, producer, partitionList, topicWithAllPartitionsOnAllRacks, partitionList, List.of(topicWithSingleRackPartitions)); + + } finally { + executor.shutdownNow(); + consumers.forEach(Consumer::close); + } + } + } + + private Map<String, Uuid> getTopicIds(ClusterInstance cluster, Collection<String> topics) throws Exception { + Map<String, Uuid> topicIds = new HashMap<>(); + try (Admin admin = cluster.admin()) { + Map<String, TopicDescription> descriptions = admin.describeTopics(topics) + .allTopicNames() + .get(); + descriptions.forEach((name, desc) -> topicIds.put(name, desc.topicId())); + return topicIds; + } + } + + private LinkedHashMap<TopicPartition, FetchRequest.PartitionData> createPartitionMap( + ClusterInstance cluster, + int maxPartitionBytes, + List<TopicPartition> topicPartitions, + Map<TopicPartition, Long> offsetMap) throws Exception { + + Map<String, Uuid> topicIds = getTopicIds(cluster, topicPartitions.stream() + .map(TopicPartition::topic) + .collect(Collectors.toSet())); + var partitionMap = new LinkedHashMap<TopicPartition, FetchRequest.PartitionData>(); + for (TopicPartition tp : topicPartitions) { + Uuid topicId = topicIds.getOrDefault(tp.topic(), Uuid.ZERO_UUID); Review Comment: Why do we use getOrDefault? The default value is illegal in this case, right? Could we use get instead? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
