KAFKA-8443 Broker support for fetch from followers

clients/src/main/java/org/apache/kafka/common/replica/LeaderReplicaSelector.java

@@ -0,0 +1,30 @@
+/*
+ * 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.replica;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Optional;
+
+public class LeaderReplicaSelector implements ReplicaSelector {
+    @Override
+    public Optional<ReplicaView> select(TopicPartition topicPartition,
+                                        ClientMetadata clientMetadata,
+                                        PartitionView partitionView) {
+        return partitionView.leader();
+    }
+}

clients/src/main/java/org/apache/kafka/common/replica/MostCaughtUpReplicaSelector.java

@@ -0,0 +1,34 @@
+/*
+ * 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.replica;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Comparator;
+import java.util.Optional;
+
+public class MostCaughtUpReplicaSelector implements ReplicaSelector {
+    @Override
+    public Optional<ReplicaView> select(TopicPartition topicPartition,
+                                        ClientMetadata clientMetadata,
+                                        PartitionView partitionView) {
+        return partitionView.replicas().stream()
+            .max(Comparator.comparing(ReplicaView::logOffset)
+                .thenComparing(ReplicaView::lastCaughtUpTimeMs)
+                .thenComparing(replicaInfo -> replicaInfo.endpoint().id()));
+    }
+}

clients/src/main/java/org/apache/kafka/common/replica/RackAwareReplicaSelector.java

@@ -0,0 +1,64 @@
+/*
+ * 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.replica;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class RackAwareReplicaSelector implements ReplicaSelector {
+
+    private final MostCaughtUpReplicaSelector tieBreaker = new MostCaughtUpReplicaSelector();
+
+    @Override
+    public Optional<ReplicaView> select(TopicPartition topicPartition,
+                                        ClientMetadata clientMetadata,
+                                        PartitionView partitionView) {
+        if (clientMetadata.rackId != null && !clientMetadata.rackId.isEmpty()) {
+            Set<ReplicaView> sameRackReplicas = partitionView.replicas().stream()
+                    .filter(replicaInfo -> clientMetadata.rackId.equalsIgnoreCase(replicaInfo.endpoint().rack()))
+                    .collect(Collectors.toSet());
+            if (sameRackReplicas.isEmpty()) {
+                return partitionView.leader();
+            } else {
+                Optional<ReplicaView> leader = partitionView.leader().filter(sameRackReplicas::contains);
+                if (leader.isPresent()) {
+                    // Use the leader if it's in this rack
+                    return leader;
+                } else {
+                    // Otherwise, get the most caught-up replica
+                    PartitionView sameRackPartition = new PartitionView() {
+                        @Override
+                        public Set<ReplicaView> replicas() {
+                            return sameRackReplicas;
+                        }
+
+                        @Override
+                        public Optional<ReplicaView> leader() {
+                            return partitionView.leader();
+                        }
+                    };
+                    return tieBreaker.select(topicPartition, clientMetadata, sameRackPartition);
+                }
+            }
+        } else {
+            return partitionView.leader();
+        }
+    }
+}

clients/src/main/java/org/apache/kafka/common/replica/ReplicaSelector.java

@@ -0,0 +1,99 @@
+/*
+ * 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.replica;
+
+import org.apache.kafka.common.Configurable;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Pluggable interface to select a preferred read replica given the current set of replicas for a partition
+ * and information from the client.
+ */
+public interface ReplicaSelector extends Configurable, Closeable {
+
+    /**
+     * Select the preferred replica a client should use for fetching. If no replica is available, this will return an
+     * empty optional.
+     */
+    Optional<ReplicaView> select(TopicPartition topicPartition,
+                                 ClientMetadata clientMetadata,
+                                 PartitionView partitionView);
+    @Override
+    default void close() throws IOException {
+        // No-op by default
+    }
+
+    @Override
+    default void configure(Map<String, ?> configs) {
+        // No-op by default
+    }
+
+    /**
+     * Holder for all the client metadata required to determine a preferred replica.
+     */
+    class ClientMetadata {
+        public static final ClientMetadata NO_METADATA = new ClientMetadata("", "", null, null, null);
+
+        public final String rackId;
+        public final String clientId;
+        public final InetAddress clientAddress;
+        public final KafkaPrincipal principal;
+        public final String listenerName;
+
+        public ClientMetadata(String rackId, String clientId, InetAddress clientAddress, KafkaPrincipal principal, String listenerName) {
+            this.rackId = rackId;
+            this.clientId = clientId;
+            this.clientAddress = clientAddress;
+            this.principal = principal;
+            this.listenerName = listenerName;
+        }
+    }
+
+    /**
+     * View of a partition used by {@link ReplicaSelector} to determine a preferred replica.
+     */
+    interface PartitionView {
+        Set<ReplicaView> replicas();
+
+        Optional<ReplicaView> leader();
+    }
+
+    /**
+     * View of a replica used by {@link ReplicaSelector} to determine a preferred replica.
+     */
+    interface ReplicaView {
+        boolean isLeader();
+
+        Node endpoint();
+
+        long logOffset();
+
+        long lastCaughtUpTimeMs();
+    }
+
+}
+
+

clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java

@@ -329,7 +329,7 @@ public int hashCode() {
             result = 31 * result + Long.hashCode(highWatermark);
             result = 31 * result + Long.hashCode(lastStableOffset);
             result = 31 * result + Long.hashCode(logStartOffset);
-            result = 31 * result + (preferredReadReplica != null ? preferredReadReplica.hashCode() : 0);
+            result = 31 * result + Objects.hashCode(preferredReadReplica);
             result = 31 * result + (abortedTransactions != null ? abortedTransactions.hashCode() : 0);
             result = 31 * result + (records != null ? records.hashCode() : 0);
             return result;

clients/src/test/java/org/apache/kafka/common/replica/ReplicaSelectorTest.java

@@ -0,0 +1,139 @@
+/*
+ * 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.replica;
+
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.kafka.test.TestUtils.assertOptional;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ReplicaSelectorTest {
+    @Test
+    public void testLeaderSelector() {
+        TopicPartition tp = new TopicPartition("test", 0);
+
+        Set<ReplicaSelector.ReplicaView> replicaViewSet = replicaInfoSet();
+        ReplicaSelector.PartitionView partitionView = partitionInfo(replicaViewSet);
+
+        ReplicaSelector selector = new LeaderReplicaSelector();
+        Optional<ReplicaSelector.ReplicaView> selected;
+
+        selected = selector.select(tp, ReplicaSelector.ClientMetadata.NO_METADATA, partitionView);
+        assertOptional(selected, replicaInfo -> {
+            assertTrue(replicaInfo.isLeader());
+            assertEquals(replicaInfo.endpoint().id(), 0);
+        });
+
+        selected = selector.select(tp, ReplicaSelector.ClientMetadata.NO_METADATA, partitionInfo(Collections.emptySet()));
+        assertFalse(selected.isPresent());
+    }
+
+    @Test
+    public void testSameRackSelector() {
+        TopicPartition tp = new TopicPartition("test", 0);
+
+        Set<ReplicaSelector.ReplicaView> replicaViewSet = replicaInfoSet();
+        ReplicaSelector.PartitionView partitionView = partitionInfo(replicaViewSet);
+
+        ReplicaSelector selector = new RackAwareReplicaSelector();
+        Optional<ReplicaSelector.ReplicaView> selected = selector.select(tp, metadata("rack-b"), partitionView);
+        assertOptional(selected, replicaInfo -> {
+            assertEquals("Expect replica to be in rack-b", replicaInfo.endpoint().rack(), "rack-b");
+            assertEquals("Expected replica 3 since it is more caught-up", replicaInfo.endpoint().id(), 3);
+        });
+
+        selected = selector.select(tp, metadata("not-a-rack"), partitionView);
+        assertOptional(selected, replicaInfo -> {
+            assertTrue("Expect leader when we can't find any nodes in given rack", replicaInfo.isLeader());
+        });
+
+        selected = selector.select(tp, metadata("rack-a"), partitionView);
+        assertOptional(selected, replicaInfo -> {
+            assertEquals("Expect replica to be in rack-a", replicaInfo.endpoint().rack(), "rack-a");
+            assertTrue("Expect the leader since it's in rack-a", replicaInfo.isLeader());
+        });
+
+
+    }
+
+    static Set<ReplicaSelector.ReplicaView> replicaInfoSet() {
+        return Stream.of(
+                replicaInfo(new Node(0, "host0", 1234, "rack-a"), true, 4, 10),
+                replicaInfo(new Node(1, "host1", 1234, "rack-a"), false, 2, 5),
+                replicaInfo(new Node(2, "host2", 1234, "rack-b"), false, 3, 7),
+                replicaInfo(new Node(3, "host3", 1234, "rack-b"), false, 4, 8)
+
+        ).collect(Collectors.toSet());
+    }
+
+    static ReplicaSelector.ReplicaView replicaInfo(Node node, boolean isLeader, long logOffset, long lastCaughtUpTimeMs) {
+        return new ReplicaSelector.ReplicaView() {
+
+            @Override
+            public boolean isLeader() {
+                return isLeader;
+            }
+
+            @Override
+            public Node endpoint() {
+                return node;
+            }
+
+            @Override
+            public long logOffset() {
+                return logOffset;
+            }
+
+            @Override
+            public long lastCaughtUpTimeMs() {
+                return lastCaughtUpTimeMs;
+            }
+        };
+    }
+
+    static ReplicaSelector.PartitionView partitionInfo(Set<ReplicaSelector.ReplicaView> replicaViewSet) {
+        return new ReplicaSelector.PartitionView() {
+            @Override
+            public Set<ReplicaSelector.ReplicaView> replicas() {
+                return replicaViewSet;
+            }
+
+            @Override
+            public Optional<ReplicaSelector.ReplicaView> leader() {
+                return replicaViewSet.stream().filter(ReplicaSelector.ReplicaView::isLeader).findFirst();
+            }
+        };
+    }
+
+    static ReplicaSelector.ClientMetadata metadata(String rack) {
+        return new ReplicaSelector.ClientMetadata(rack, "test-client",
+                InetAddress.getLoopbackAddress(), KafkaPrincipal.ANONYMOUS, "test");
+
+    }
+}