diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java index 1d71ea17c0d61..a7d346aa6eef5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java +++ b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java @@ -100,8 +100,8 @@ public static ChannelBuilder createChannelBuilder(AbstractConfig config, Time ti clientSaslMechanism, time, true); } - static List resolve(String host, ClientDnsLookup clientDnsLookup) throws UnknownHostException { - InetAddress[] addresses = InetAddress.getAllByName(host); + static List resolve(String host, ClientDnsLookup clientDnsLookup, HostResolver hostResolver) throws UnknownHostException { + InetAddress[] addresses = hostResolver.resolve(host); if (ClientDnsLookup.USE_ALL_DNS_IPS == clientDnsLookup) { return filterPreferredAddresses(addresses); } else { diff --git a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java index ec3465ac4395b..c1461fae6fe0c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java +++ b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java @@ -40,13 +40,15 @@ final class ClusterConnectionStates { private final double reconnectBackoffMaxExp; private final Map nodeState; private final Logger log; + private final HostResolver hostResolver; - public ClusterConnectionStates(long reconnectBackoffMs, long reconnectBackoffMaxMs, LogContext logContext) { + public ClusterConnectionStates(long reconnectBackoffMs, long reconnectBackoffMaxMs, LogContext logContext, HostResolver hostResolver) { this.log = logContext.logger(ClusterConnectionStates.class); this.reconnectBackoffInitMs = reconnectBackoffMs; this.reconnectBackoffMaxMs = reconnectBackoffMaxMs; this.reconnectBackoffMaxExp = Math.log(this.reconnectBackoffMaxMs / (double) Math.max(reconnectBackoffMs, 1)) / Math.log(RECONNECT_BACKOFF_EXP_BASE); this.nodeState = new HashMap<>(); + this.hostResolver = hostResolver; } /** @@ -139,7 +141,7 @@ public void connecting(String id, long now, String host, ClientDnsLookup clientD // Create a new NodeConnectionState if nodeState does not already contain one // for the specified id or if the hostname associated with the node id changed. nodeState.put(id, new NodeConnectionState(ConnectionState.CONNECTING, now, - this.reconnectBackoffInitMs, host, clientDnsLookup)); + this.reconnectBackoffInitMs, host, clientDnsLookup, hostResolver)); } /** @@ -373,9 +375,10 @@ private static class NodeConnectionState { private int addressIndex; private final String host; private final ClientDnsLookup clientDnsLookup; + private final HostResolver hostResolver; private NodeConnectionState(ConnectionState state, long lastConnectAttempt, long reconnectBackoffMs, - String host, ClientDnsLookup clientDnsLookup) { + String host, ClientDnsLookup clientDnsLookup, HostResolver hostResolver) { this.state = state; this.addresses = Collections.emptyList(); this.addressIndex = -1; @@ -386,6 +389,7 @@ private NodeConnectionState(ConnectionState state, long lastConnectAttempt, long this.throttleUntilTimeMs = 0; this.host = host; this.clientDnsLookup = clientDnsLookup; + this.hostResolver = hostResolver; } public String host() { @@ -400,7 +404,7 @@ public String host() { private InetAddress currentAddress() throws UnknownHostException { if (addresses.isEmpty()) { // (Re-)initialize list - addresses = ClientUtils.resolve(host, clientDnsLookup); + addresses = ClientUtils.resolve(host, clientDnsLookup, hostResolver); addressIndex = 0; } diff --git a/clients/src/main/java/org/apache/kafka/clients/DefaultHostResolver.java b/clients/src/main/java/org/apache/kafka/clients/DefaultHostResolver.java new file mode 100644 index 0000000000000..786173e3a2751 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/DefaultHostResolver.java @@ -0,0 +1,29 @@ +/* + * 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.clients; + +import java.net.InetAddress; +import java.net.UnknownHostException; + +public class DefaultHostResolver implements HostResolver { + + @Override + public InetAddress[] resolve(String host) throws UnknownHostException { + return InetAddress.getAllByName(host); + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/HostResolver.java b/clients/src/main/java/org/apache/kafka/clients/HostResolver.java new file mode 100644 index 0000000000000..80209ca8a9476 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/HostResolver.java @@ -0,0 +1,26 @@ +/* + * 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.clients; + +import java.net.InetAddress; +import java.net.UnknownHostException; + +public interface HostResolver { + + InetAddress[] resolve(String host) throws UnknownHostException; +} diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java index 047b03585c018..0882e702537df 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java @@ -139,9 +139,8 @@ public NetworkClient(Selectable selector, boolean discoverBrokerVersions, ApiVersions apiVersions, LogContext logContext) { - this(null, + this(selector, metadata, - selector, clientId, maxInFlightRequestsPerConnection, reconnectBackoffMs, @@ -187,7 +186,8 @@ public NetworkClient(Selectable selector, discoverBrokerVersions, apiVersions, throttleTimeSensor, - logContext); + logContext, + new DefaultHostResolver()); } public NetworkClient(Selectable selector, @@ -219,7 +219,8 @@ public NetworkClient(Selectable selector, discoverBrokerVersions, apiVersions, null, - logContext); + logContext, + new DefaultHostResolver()); } private NetworkClient(MetadataUpdater metadataUpdater, @@ -237,7 +238,8 @@ private NetworkClient(MetadataUpdater metadataUpdater, boolean discoverBrokerVersions, ApiVersions apiVersions, Sensor throttleTimeSensor, - LogContext logContext) { + LogContext logContext, + HostResolver hostResolver) { /* It would be better if we could pass `DefaultMetadataUpdater` from the public constructor, but it's not * possible because `DefaultMetadataUpdater` is an inner class and it can only be instantiated after the * super constructor is invoked. @@ -252,7 +254,7 @@ private NetworkClient(MetadataUpdater metadataUpdater, this.selector = selector; this.clientId = clientId; this.inFlightRequests = new InFlightRequests(maxInFlightRequestsPerConnection); - this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax, logContext); + this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax, logContext, hostResolver); this.socketSendBuffer = socketSendBuffer; this.socketReceiveBuffer = socketReceiveBuffer; this.correlation = 0; diff --git a/clients/src/test/java/org/apache/kafka/clients/AddressChangeHostResolver.java b/clients/src/test/java/org/apache/kafka/clients/AddressChangeHostResolver.java new file mode 100644 index 0000000000000..f2c17a2b85836 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/AddressChangeHostResolver.java @@ -0,0 +1,39 @@ +/* + * 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.clients; + +import java.net.InetAddress; + +class AddressChangeHostResolver implements HostResolver { + private boolean useNewAddresses; + private InetAddress[] initialAddresses; + private InetAddress[] newAddresses; + + public AddressChangeHostResolver(InetAddress[] initialAddresses, InetAddress[] newAddresses) { + this.initialAddresses = initialAddresses; + this.newAddresses = newAddresses; + } + + @Override + public InetAddress[] resolve(String host) { + return useNewAddresses ? newAddresses : initialAddresses; + } + + public void changeAddresses() { + useNewAddresses = true; + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java b/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java index 99f97e191e295..735804e0f321a 100644 --- a/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java @@ -31,9 +31,10 @@ public class ClientUtilsTest { + private HostResolver hostResolver = new DefaultHostResolver(); @Test - public void testParseAndValidateAddresses() throws UnknownHostException { + public void testParseAndValidateAddresses() { checkWithoutLookup("127.0.0.1:8000"); checkWithoutLookup("localhost:8080"); checkWithoutLookup("[::1]:8000"); @@ -98,17 +99,17 @@ public void testFilterPreferredAddresses() throws UnknownHostException { @Test(expected = UnknownHostException.class) public void testResolveUnknownHostException() throws UnknownHostException { - ClientUtils.resolve("some.invalid.hostname.foo.bar.local", ClientDnsLookup.DEFAULT); + ClientUtils.resolve("some.invalid.hostname.foo.bar.local", ClientDnsLookup.DEFAULT, hostResolver); } @Test public void testResolveDnsLookup() throws UnknownHostException { - assertEquals(1, ClientUtils.resolve("localhost", ClientDnsLookup.DEFAULT).size()); + assertEquals(1, ClientUtils.resolve("localhost", ClientDnsLookup.DEFAULT, hostResolver).size()); } @Test public void testResolveDnsLookupAllIps() throws UnknownHostException { - assertTrue(ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); + assertTrue(ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.USE_ALL_DNS_IPS, hostResolver).size() > 1); } private List checkWithoutLookup(String... url) { diff --git a/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java b/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java index fbb449715389f..5a21c3881ea32 100644 --- a/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java @@ -23,11 +23,12 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; -import java.lang.reflect.Field; -import java.lang.reflect.Method; import java.net.InetAddress; import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.Arrays; import org.apache.kafka.common.errors.AuthenticationException; import org.apache.kafka.common.utils.LogContext; @@ -37,19 +38,45 @@ public class ClusterConnectionStatesTest { + private static ArrayList initialAddresses; + private static ArrayList newAddresses; + + static { + try { + initialAddresses = new ArrayList<>(Arrays.asList( + InetAddress.getByName("10.200.20.100"), + InetAddress.getByName("10.200.20.101"), + InetAddress.getByName("10.200.20.102") + )); + newAddresses = new ArrayList<>(Arrays.asList( + InetAddress.getByName("10.200.20.103"), + InetAddress.getByName("10.200.20.104"), + InetAddress.getByName("10.200.20.105") + )); + } catch (UnknownHostException e) { + fail("Attempted to create an invalid InetAddress, this should not happen"); + } + } + private final MockTime time = new MockTime(); private final long reconnectBackoffMs = 10 * 1000; private final long reconnectBackoffMax = 60 * 1000; private final double reconnectBackoffJitter = 0.2; private final String nodeId1 = "1001"; private final String nodeId2 = "2002"; - private final String hostTwoIps = "kafka.apache.org"; - + private final String hostTwoIps = "multiple.ip.address"; private ClusterConnectionStates connectionStates; + // For testing nodes with a single IP address, use localhost and default DNS resolution + private DefaultHostResolver singleIPHostResolver = new DefaultHostResolver(); + + // For testing nodes with multiple IP addresses, mock DNS resolution to get consistent results + private AddressChangeHostResolver multipleIPHostResolver = new AddressChangeHostResolver( + initialAddresses.toArray(new InetAddress[0]), newAddresses.toArray(new InetAddress[0])); + @Before public void setup() { - this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax, new LogContext()); + this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax, new LogContext(), this.singleIPHostResolver); } @Test @@ -246,7 +273,7 @@ public void testSingleIPWithDefault() throws UnknownHostException { @Test public void testSingleIPWithUseAll() throws UnknownHostException { - assertEquals(1, ClientUtils.resolve("localhost", ClientDnsLookup.USE_ALL_DNS_IPS).size()); + assertEquals(1, ClientUtils.resolve("localhost", ClientDnsLookup.USE_ALL_DNS_IPS, singleIPHostResolver).size()); connectionStates.connecting(nodeId1, time.milliseconds(), "localhost", ClientDnsLookup.USE_ALL_DNS_IPS); InetAddress currAddress = connectionStates.currentAddress(nodeId1); @@ -256,7 +283,9 @@ public void testSingleIPWithUseAll() throws UnknownHostException { @Test public void testMultipleIPsWithDefault() throws UnknownHostException { - assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); + setupMultipleIPs(); + + assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS, multipleIPHostResolver).size() > 1); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.DEFAULT); InetAddress currAddress = connectionStates.currentAddress(nodeId1); @@ -266,7 +295,9 @@ public void testMultipleIPsWithDefault() throws UnknownHostException { @Test public void testMultipleIPsWithUseAll() throws UnknownHostException { - assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); + setupMultipleIPs(); + + assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS, multipleIPHostResolver).size() > 1); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS); InetAddress addr1 = connectionStates.currentAddress(nodeId1); @@ -279,20 +310,15 @@ public void testMultipleIPsWithUseAll() throws UnknownHostException { } @Test - public void testHostResolveChange() throws UnknownHostException, ReflectiveOperationException { - assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1); + public void testHostResolveChange() throws UnknownHostException { + setupMultipleIPs(); + + assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS, multipleIPHostResolver).size() > 1); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.DEFAULT); InetAddress addr1 = connectionStates.currentAddress(nodeId1); - // reflection to simulate host change in DNS lookup - Method nodeStateMethod = connectionStates.getClass().getDeclaredMethod("nodeState", String.class); - nodeStateMethod.setAccessible(true); - Object nodeState = nodeStateMethod.invoke(connectionStates, nodeId1); - Field hostField = nodeState.getClass().getDeclaredField("host"); - hostField.setAccessible(true); - hostField.set(nodeState, "localhost"); - + multipleIPHostResolver.changeAddresses(); connectionStates.connecting(nodeId1, time.milliseconds(), "localhost", ClientDnsLookup.DEFAULT); InetAddress addr2 = connectionStates.currentAddress(nodeId1); @@ -301,9 +327,12 @@ public void testHostResolveChange() throws UnknownHostException, ReflectiveOpera @Test public void testNodeWithNewHostname() throws UnknownHostException { + setupMultipleIPs(); + connectionStates.connecting(nodeId1, time.milliseconds(), "localhost", ClientDnsLookup.DEFAULT); InetAddress addr1 = connectionStates.currentAddress(nodeId1); + this.multipleIPHostResolver.changeAddresses(); connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.DEFAULT); InetAddress addr2 = connectionStates.currentAddress(nodeId1); @@ -320,4 +349,8 @@ public void testIsPreparingConnection() { connectionStates.disconnected(nodeId1, time.milliseconds()); assertFalse(connectionStates.isPreparingConnection(nodeId1)); } + + private void setupMultipleIPs() { + this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax, new LogContext(), this.multipleIPHostResolver); + } }