diff --git a/src/main/java/org/cache/Main.java b/src/main/java/org/cache/Main.java index 7df0af4..4260ff5 100644 --- a/src/main/java/org/cache/Main.java +++ b/src/main/java/org/cache/Main.java @@ -1,6 +1,7 @@ package org.cache; import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterHealthMonitor; import org.cache.cluster.routing.ClusterForwardingClient; import org.cache.cluster.routing.RoutedCacheService; import org.cache.config.CacheConfig; @@ -40,7 +41,7 @@ public static void main(String[] args) { configuredCacheConfig = loadConfiguration(); var app = new SpringApplication(Main.class); app.setDefaultProperties(Map.of( - "server.port", configuredCacheConfig.cacheNode().httpPort() + "server.port", configuredCacheConfig.cacheNode().getHttpPort() )); app.run(args); } @@ -139,6 +140,15 @@ public ClusterForwardingClient clusterForwardingClient() { return new ClusterForwardingClient(); } + @Bean + public ClusterHealthMonitor clusterHealthMonitor( + CacheNode cacheNode, + CacheConfig cacheConfig, + ClusterForwardingClient forwardingClient + ) { + return new ClusterHealthMonitor(cacheNode, cacheConfig.clusterInfo(), forwardingClient); + } + @Bean(destroyMethod = "shutdownNow") public ExecutorService tcpClientExecutor() { return Executors.newFixedThreadPool(SERVER_THREAD_COUNT); @@ -156,7 +166,7 @@ public TcpCacheServer clientTcpCacheServer( @Qualifier("commandProcessor") CommandProcessor commandProcessor ) throws IOException { return new TcpCacheServer( - new ServerSocket(cacheNode.tcpPort()), + new ServerSocket(cacheNode.getTcpPort()), tcpClientExecutor, socket -> new ClientConnectionHandler(socket, commandProcessor::process) ); @@ -169,7 +179,7 @@ public TcpCacheServer clusterTcpCacheServer( @Qualifier("clusterCommandProcessor") CommandProcessor clusterCommandProcessor ) throws IOException { return new TcpCacheServer( - new ServerSocket(cacheNode.clusterPort()), + new ServerSocket(cacheNode.getClusterPort()), clusterClientExecutor, socket -> new ClientConnectionHandler(socket, clusterCommandProcessor::process) ); diff --git a/src/main/java/org/cache/cluster/CacheNode.java b/src/main/java/org/cache/cluster/CacheNode.java index 440c8f7..367082a 100644 --- a/src/main/java/org/cache/cluster/CacheNode.java +++ b/src/main/java/org/cache/cluster/CacheNode.java @@ -1,15 +1,83 @@ package org.cache.cluster; -public record CacheNode( - String id, - String host, - int httpPort, - int tcpPort, - int clusterPort, - NodeStatus status -) { +import java.util.Objects; + +public final class CacheNode { + + private final String id; + + private final String host; + + private final int httpPort; + + private final int tcpPort; + + private final int clusterPort; + + private volatile NodeStatus status; public CacheNode(String id, String host, int httpPort, int tcpPort, int clusterPort) { this(id, host, httpPort, tcpPort, clusterPort, NodeStatus.HEALTHY); } + + public CacheNode(String id, String host, int httpPort, int tcpPort, int clusterPort, NodeStatus status) { + this.id = Objects.requireNonNull(id, "Node id must not be null"); + this.host = Objects.requireNonNull(host, "Node host must not be null"); + this.httpPort = httpPort; + this.tcpPort = tcpPort; + this.clusterPort = clusterPort; + this.status = Objects.requireNonNull(status, "Node status must not be null"); + } + + public String getId() { + return id; + } + + public String getHost() { + return host; + } + + public int getHttpPort() { + return httpPort; + } + + public int getTcpPort() { + return tcpPort; + } + + public int getClusterPort() { + return clusterPort; + } + + public NodeStatus getStatus() { + return status; + } + + public void setStatus(NodeStatus status) { + this.status = Objects.requireNonNull(status, "Node status must not be null"); + } + + @Override + public boolean equals(Object object) { + if (this == object) { + return true; + } + + if (!(object instanceof CacheNode cacheNode)) { + return false; + } + + return httpPort == cacheNode.httpPort + && tcpPort == cacheNode.tcpPort + && clusterPort == cacheNode.clusterPort + && id.equals(cacheNode.id) + && host.equals(cacheNode.host) + && status == cacheNode.status; + } + + @Override + public int hashCode() { + return Objects.hash(id, host, httpPort, tcpPort, clusterPort, status); + } + } diff --git a/src/main/java/org/cache/cluster/ClusterHealthMonitor.java b/src/main/java/org/cache/cluster/ClusterHealthMonitor.java new file mode 100644 index 0000000..a502f62 --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterHealthMonitor.java @@ -0,0 +1,115 @@ +package org.cache.cluster; + +import org.cache.cluster.routing.ClusterForwardingClient; +import org.springframework.context.SmartLifecycle; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +public class ClusterHealthMonitor implements SmartLifecycle { + + private static final int SUSPECTED_FAILURE_COUNT = 2; + private static final int UNAVAILABLE_FAILURE_COUNT = 3; + private static final long INITIAL_DELAY_SECONDS = 5; + private static final long CHECK_INTERVAL_SECONDS = 5; + + private final CacheNode currentNode; + private final ClusterInfo clusterInfo; + private final ClusterForwardingClient forwardingClient; + private final Map failureCounts = new HashMap<>(); + private final ScheduledExecutorService executor; + private volatile boolean running; + + public ClusterHealthMonitor( + CacheNode currentNode, + ClusterInfo clusterInfo, + ClusterForwardingClient forwardingClient + ) { + this(currentNode, clusterInfo, forwardingClient, Executors.newSingleThreadScheduledExecutor()); + } + + ClusterHealthMonitor( + CacheNode currentNode, + ClusterInfo clusterInfo, + ClusterForwardingClient forwardingClient, + ScheduledExecutorService executor + ) { + this.currentNode = currentNode; + this.clusterInfo = clusterInfo; + this.forwardingClient = forwardingClient; + this.executor = executor; + } + + @Override + public void start() { + if (running || clusterInfo == null) { + return; + } + + running = true; + executor.scheduleAtFixedRate( + this::checkClusterSafely, + INITIAL_DELAY_SECONDS, + CHECK_INTERVAL_SECONDS, + TimeUnit.SECONDS + ); + } + + @Override + public void stop() { + running = false; + executor.shutdownNow(); + } + + @Override + public boolean isRunning() { + return running; + } + + void checkCluster() { + if (clusterInfo == null) { + return; + } + + for (CacheNode node : clusterInfo.nodes()) { + if (!node.getId().equals(currentNode.getId())) { + checkNode(node); + } + } + } + + private void checkClusterSafely() { + try { + checkCluster(); + } catch (RuntimeException exception) { + System.err.println("Cluster health check failed: " + exception.getMessage()); + } + } + + private void checkNode(CacheNode node) { + if (forwardingClient.ping(node)) { + failureCounts.remove(node.getId()); + node.setStatus(NodeStatus.HEALTHY); + return; + } + + int failures = failureCounts.getOrDefault(node.getId(), 0) + 1; + failureCounts.put(node.getId(), failures); + node.setStatus(statusFor(failures)); + } + + private NodeStatus statusFor(int failures) { + if (failures >= UNAVAILABLE_FAILURE_COUNT) { + return NodeStatus.UNAVAILABLE; + } + + if (failures == SUSPECTED_FAILURE_COUNT) { + return NodeStatus.SUSPECTED; + } + + return NodeStatus.HEALTHY; + } +} diff --git a/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java b/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java index 732de90..accc031 100644 --- a/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java +++ b/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java @@ -2,6 +2,7 @@ import org.cache.cluster.CacheNode; import org.cache.cluster.ClusterInfo; +import org.cache.cluster.NodeStatus; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; @@ -40,7 +41,12 @@ public ConsistentHashRing(ClusterInfo clusterInfo, int virtualNodeCount) { } public CacheNode nodeFor(String key) { - return nodesFor(key).getFirst(); + List nodes = nodesFor(key); + if (nodes.isEmpty()) { + throw new IllegalStateException("No available nodes in hash ring"); + } + + return nodes.getFirst(); } public List nodesFor(String key) { @@ -64,13 +70,13 @@ private void buildRing(List nodes, int virtualNodeCount) { private void addNode(CacheNode node, int virtualNodeCount) { for (int index = 0; index < virtualNodeCount; index++) { - ring.put(hash(node.id() + "#" + index), node); + ring.put(hash(node.getId() + "#" + index), node); } } private void addNodesFromRing(long startHash, List nodes, Set selectedNodeIds) { for (CacheNode node : ring.tailMap(startHash, true).values()) { - if (selectedNodeIds.add(node.id())) { + if (node.getStatus() != NodeStatus.UNAVAILABLE && selectedNodeIds.add(node.getId())) { nodes.add(node); } diff --git a/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java b/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java index 529882d..71b0ac6 100644 --- a/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java +++ b/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java @@ -8,6 +8,9 @@ public class ClusterForwardingClient { + private static final List PING_COMMAND = List.of("PING"); + private static final List PONG_RESPONSE = List.of("PONG"); + private final RespCommandClient commandClient; public ClusterForwardingClient() { @@ -20,9 +23,17 @@ public ClusterForwardingClient() { public List forward(CacheNode targetNode, List commandParts) { try { - return commandClient.send(targetNode.host(), targetNode.clusterPort(), commandParts); + return commandClient.send(targetNode.getHost(), targetNode.getClusterPort(), commandParts); + } catch (IOException exception) { + throw new ClusterForwardingException("Failed to forward request to node: " + targetNode.getId(), exception); + } + } + + public boolean ping(CacheNode targetNode) { + try { + return PONG_RESPONSE.equals(commandClient.send(targetNode.getHost(), targetNode.getClusterPort(), PING_COMMAND)); } catch (IOException exception) { - throw new ClusterForwardingException("Failed to forward request to node: " + targetNode.id(), exception); + return false; } } } diff --git a/src/main/java/org/cache/cluster/routing/RoutedCacheService.java b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java index 837b3d7..34bc0e7 100644 --- a/src/main/java/org/cache/cluster/routing/RoutedCacheService.java +++ b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java @@ -173,7 +173,7 @@ private ReplicationTargets writeTargetsFor(K key) { if (!includesCurrentNode && !forwardingAllowed) { throw new ClusterForwardingException("Request routed to wrong node. Expected one of owners: " - + ownerIds(owners) + ", current node: " + currentNode.id()); + + ownerIds(owners) + ", current node: " + currentNode.getId()); } if (!forwardingAllowed) { @@ -199,7 +199,7 @@ private List readOwnersFor(K key) { if (!includesCurrentNode && !forwardingAllowed) { throw new ClusterForwardingException("Request routed to wrong node. Expected one of owners: " - + ownerIds(owners) + ", current node: " + currentNode.id()); + + ownerIds(owners) + ", current node: " + currentNode.getId()); } if (!forwardingAllowed) { @@ -245,12 +245,12 @@ private Optional> remoteLrange(CacheNode owner, K key, int from, in } private boolean isCurrentNode(CacheNode owner) { - return owner.id().equals(currentNode.id()); + return owner.getId().equals(currentNode.getId()); } private String ownerIds(List owners) { return owners.stream() - .map(CacheNode::id) + .map(CacheNode::getId) .toList() .toString(); } diff --git a/src/main/java/org/cache/config/CacheConfigLoader.java b/src/main/java/org/cache/config/CacheConfigLoader.java index de1787d..94c92d7 100644 --- a/src/main/java/org/cache/config/CacheConfigLoader.java +++ b/src/main/java/org/cache/config/CacheConfigLoader.java @@ -140,13 +140,13 @@ private void validateClusterInfo(int replicationFactor, List nodes) { Set hostPorts = new HashSet<>(); for (CacheNode node : nodes) { - if (!nodeIds.add(node.id())) { - throw new CacheConfigException("Cluster node ids must be unique: " + node.id()); + if (!nodeIds.add(node.getId())) { + throw new CacheConfigException("Cluster node ids must be unique: " + node.getId()); } - addHostPort(hostPorts, node.host(), node.httpPort()); - addHostPort(hostPorts, node.host(), node.tcpPort()); - addHostPort(hostPorts, node.host(), node.clusterPort()); + addHostPort(hostPorts, node.getHost(), node.getHttpPort()); + addHostPort(hostPorts, node.getHost(), node.getTcpPort()); + addHostPort(hostPorts, node.getHost(), node.getClusterPort()); } } diff --git a/src/test/java/org/cache/cluster/ClusterHealthMonitorTest.java b/src/test/java/org/cache/cluster/ClusterHealthMonitorTest.java new file mode 100644 index 0000000..1b3dc05 --- /dev/null +++ b/src/test/java/org/cache/cluster/ClusterHealthMonitorTest.java @@ -0,0 +1,79 @@ +package org.cache.cluster; + +import org.cache.cluster.routing.ClusterForwardingClient; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.concurrent.ScheduledExecutorService; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class ClusterHealthMonitorTest { + + private final CacheNode nodeA = node("node-a", 8080, 2020, 10001); + private final CacheNode nodeB = node("node-b", 8081, 2021, 10002); + private final ClusterInfo clusterInfo = new ClusterInfo(1, List.of(nodeA, nodeB)); + + @Test + void checkClusterMovesNodeThroughFailureStatesAfterConsecutiveFailures() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterHealthMonitor monitor = monitor(forwardingClient); + + when(forwardingClient.ping(nodeB)).thenReturn(false); + + monitor.checkCluster(); + assertEquals(NodeStatus.HEALTHY, nodeB.getStatus()); + + monitor.checkCluster(); + assertEquals(NodeStatus.SUSPECTED, nodeB.getStatus()); + + monitor.checkCluster(); + assertEquals(NodeStatus.UNAVAILABLE, nodeB.getStatus()); + } + + @Test + void checkClusterRestoresHealthyStatusAfterSuccessfulPing() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterHealthMonitor monitor = monitor(forwardingClient); + + when(forwardingClient.ping(nodeB)).thenReturn(false, false, true); + + monitor.checkCluster(); + monitor.checkCluster(); + assertEquals(NodeStatus.SUSPECTED, nodeB.getStatus()); + + monitor.checkCluster(); + assertEquals(NodeStatus.HEALTHY, nodeB.getStatus()); + } + + @Test + void checkClusterDoesNothingWhenClusterIsDisabled() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterHealthMonitor monitor = new ClusterHealthMonitor( + nodeA, + null, + forwardingClient, + mock(ScheduledExecutorService.class) + ); + + monitor.checkCluster(); + + verifyNoInteractions(forwardingClient); + } + + private ClusterHealthMonitor monitor(ClusterForwardingClient forwardingClient) { + return new ClusterHealthMonitor( + nodeA, + clusterInfo, + forwardingClient, + mock(ScheduledExecutorService.class) + ); + } + + private CacheNode node(String id, int httpPort, int tcpPort, int clusterPort) { + return new CacheNode(id, "localhost", httpPort, tcpPort, clusterPort); + } +} diff --git a/src/test/java/org/cache/cluster/hashing/ConsistentHashRingTest.java b/src/test/java/org/cache/cluster/hashing/ConsistentHashRingTest.java index c0766fd..edcb06f 100644 --- a/src/test/java/org/cache/cluster/hashing/ConsistentHashRingTest.java +++ b/src/test/java/org/cache/cluster/hashing/ConsistentHashRingTest.java @@ -2,11 +2,13 @@ import org.cache.cluster.CacheNode; import org.cache.cluster.ClusterInfo; +import org.cache.cluster.NodeStatus; import org.junit.jupiter.api.Test; import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -29,7 +31,7 @@ void nodesForReturnsDistinctReplicas() { List nodes = ring.nodesFor("account:42"); assertEquals(2, nodes.size()); - assertNotEquals(nodes.get(0).id(), nodes.get(1).id()); + assertNotEquals(nodes.get(0).getId(), nodes.get(1).getId()); } @Test @@ -41,6 +43,19 @@ void nodesForWrapsAroundRing() { assertEquals(3, nodes.size()); } + @Test + void nodesForIgnoresUnavailableNodes() { + ClusterInfo clusterInfo = clusterInfo(3); + CacheNode unavailableNode = clusterInfo.nodes().get(1); + unavailableNode.setStatus(NodeStatus.UNAVAILABLE); + ConsistentHashRing ring = new ConsistentHashRing(clusterInfo, 32); + + List nodes = ring.nodesFor("account:42"); + + assertEquals(2, nodes.size()); + assertFalse(nodes.stream().anyMatch(node -> node.getId().equals(unavailableNode.getId()))); + } + @Test void constructorRejectsInvalidVirtualNodeCount() { var exception = assertThrows( diff --git a/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java b/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java index 5fdf01c..f91e9c6 100644 --- a/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java +++ b/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java @@ -8,7 +8,9 @@ import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -37,4 +39,35 @@ void forwardWrapsIoFailure() throws Exception { assertThrows(ClusterForwardingException.class, () -> forwardingClient.forward(node, List.of("GET", "fruit"))); } + + @Test + void pingReturnsTrueWhenNodeRespondsWithPong() throws Exception { + RespCommandClient commandClient = mock(RespCommandClient.class); + ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); + + when(commandClient.send("localhost", 10001, List.of("PING"))).thenReturn(List.of("PONG")); + + assertTrue(forwardingClient.ping(node)); + verify(commandClient).send("localhost", 10001, List.of("PING")); + } + + @Test + void pingReturnsFalseWhenNodeDoesNotRespondWithPong() throws Exception { + RespCommandClient commandClient = mock(RespCommandClient.class); + ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); + + when(commandClient.send("localhost", 10001, List.of("PING"))).thenReturn(List.of("ERROR", "broken")); + + assertFalse(forwardingClient.ping(node)); + } + + @Test + void pingReturnsFalseWhenConnectionFails() throws Exception { + RespCommandClient commandClient = mock(RespCommandClient.class); + ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); + + when(commandClient.send("localhost", 10001, List.of("PING"))).thenThrow(new IOException("closed")); + + assertFalse(forwardingClient.ping(node)); + } } diff --git a/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java index 8ba76f3..64a8319 100644 --- a/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java +++ b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java @@ -119,7 +119,7 @@ void getStringRejectsRemoteOwnerWhenForwardingIsDisabled() { String key = keyNotReplicatedTo(nodeA); List ownerIds = replicaOwnersFor(key) .stream() - .map(CacheNode::id) + .map(CacheNode::getId) .toList(); RoutedCacheService service = new RoutedCacheService<>( localService, @@ -144,7 +144,7 @@ void putStringWritesToLocalAndRemoteReplicas() { String key = keyReplicatedIncluding(nodeA); List remoteOwners = replicaOwnersFor(key) .stream() - .filter(owner -> !owner.id().equals(nodeA.id())) + .filter(owner -> !owner.getId().equals(nodeA.getId())) .toList(); RoutedCacheService service = new RoutedCacheService<>( localService, @@ -216,7 +216,7 @@ void putStringRejectsNonReplicaOwnerWhenForwardingIsDisabled() { String key = keyNotReplicatedTo(nodeA); List ownerIds = replicaOwnersFor(key) .stream() - .map(CacheNode::id) + .map(CacheNode::getId) .toList(); RoutedCacheService service = new RoutedCacheService<>( localService, @@ -300,12 +300,12 @@ private String keyOwnedBy(CacheNode owner) { for (int index = 0; index < 10_000; index++) { String key = "key-" + index; - if (ring.nodeFor(key).id().equals(owner.id())) { + if (ring.nodeFor(key).getId().equals(owner.getId())) { return key; } } - throw new IllegalStateException("Could not find key owned by node: " + owner.id()); + throw new IllegalStateException("Could not find key owned by node: " + owner.getId()); } private String keyReplicatedTo(CacheNode firstOwner, CacheNode secondOwner) { @@ -315,54 +315,54 @@ private String keyReplicatedTo(CacheNode firstOwner, CacheNode secondOwner) { String key = "replica-key-" + index; List ownerIds = ring.nodesFor(key) .stream() - .map(CacheNode::id) + .map(CacheNode::getId) .toList(); - if (ownerIds.equals(List.of(firstOwner.id(), secondOwner.id()))) { + if (ownerIds.equals(List.of(firstOwner.getId(), secondOwner.getId()))) { return key; } } throw new IllegalStateException("Could not find key replicated to owners: " - + firstOwner.id() + ", " + secondOwner.id()); + + firstOwner.getId() + ", " + secondOwner.getId()); } private String keyReplicatedIncluding(CacheNode owner) { for (int index = 0; index < 10_000; index++) { String key = "replica-key-" + index; boolean containsOwner = replicaOwnersFor(key).stream() - .anyMatch(replicaOwner -> replicaOwner.id().equals(owner.id())); + .anyMatch(replicaOwner -> replicaOwner.getId().equals(owner.getId())); if (containsOwner) { return key; } } - throw new IllegalStateException("Could not find key replicated to owner: " + owner.id()); + throw new IllegalStateException("Could not find key replicated to owner: " + owner.getId()); } private String keyNotReplicatedTo(CacheNode owner) { for (int index = 0; index < 10_000; index++) { String key = "replica-key-" + index; boolean containsOwner = replicaOwnersFor(key).stream() - .anyMatch(replicaOwner -> replicaOwner.id().equals(owner.id())); + .anyMatch(replicaOwner -> replicaOwner.getId().equals(owner.getId())); if (!containsOwner) { return key; } } - throw new IllegalStateException("Could not find key not replicated to owner: " + owner.id()); + throw new IllegalStateException("Could not find key not replicated to owner: " + owner.getId()); } private String keyReplicatedToRemotePrimaryAndLocalReplica() { for (int index = 0; index < 10_000; index++) { String key = "replica-key-" + index; List owners = replicaOwnersFor(key); - boolean hasRemotePrimary = !owners.getFirst().id().equals(nodeA.id()); + boolean hasRemotePrimary = !owners.getFirst().getId().equals(nodeA.getId()); boolean hasLocalReplica = owners.stream() .skip(1) - .anyMatch(owner -> owner.id().equals(nodeA.id())); + .anyMatch(owner -> owner.getId().equals(nodeA.getId())); if (hasRemotePrimary && hasLocalReplica) { return key; diff --git a/src/test/java/org/cache/config/CacheConfigLoaderTest.java b/src/test/java/org/cache/config/CacheConfigLoaderTest.java index 92c7eca..25512d5 100644 --- a/src/test/java/org/cache/config/CacheConfigLoaderTest.java +++ b/src/test/java/org/cache/config/CacheConfigLoaderTest.java @@ -22,7 +22,7 @@ void loadUsesConfigFileFromSystemProperty() { try { CacheConfig config = new CacheConfigLoader().load(); - assertEquals("node-test", config.cacheNode().id()); + assertEquals("node-test", config.cacheNode().getId()); assertEquals(250, config.capacity()); } finally { if (previousConfigFile == null) { @@ -41,11 +41,11 @@ void loadUsesDefaultsWhenConfigFileDoesNotExist() { assertEquals(0, config.defaultTtlMillis()); assertInstanceOf(StringKeyCodec.class, config.keyCodec()); assertInstanceOf(LruEvictionPolicy.class, config.evictionPolicy()); - assertEquals("node-a", config.cacheNode().id()); - assertEquals("localhost", config.cacheNode().host()); - assertEquals(8080, config.cacheNode().httpPort()); - assertEquals(2020, config.cacheNode().tcpPort()); - assertEquals(10001, config.cacheNode().clusterPort()); + assertEquals("node-a", config.cacheNode().getId()); + assertEquals("localhost", config.cacheNode().getHost()); + assertEquals(8080, config.cacheNode().getHttpPort()); + assertEquals(2020, config.cacheNode().getTcpPort()); + assertEquals(10001, config.cacheNode().getClusterPort()); assertNull(config.clusterInfo()); } @@ -58,11 +58,11 @@ void loadReadsConfiguredValues() { assertInstanceOf(IntegerKeyCodec.class, config.keyCodec()); assertEquals(42, config.keyCodec().decode("42")); assertInstanceOf(MruEvictionPolicy.class, config.evictionPolicy()); - assertEquals("node-test", config.cacheNode().id()); - assertEquals("127.0.0.1", config.cacheNode().host()); - assertEquals(18080, config.cacheNode().httpPort()); - assertEquals(12020, config.cacheNode().tcpPort()); - assertEquals(11001, config.cacheNode().clusterPort()); + assertEquals("node-test", config.cacheNode().getId()); + assertEquals("127.0.0.1", config.cacheNode().getHost()); + assertEquals(18080, config.cacheNode().getHttpPort()); + assertEquals(12020, config.cacheNode().getTcpPort()); + assertEquals(11001, config.cacheNode().getClusterPort()); assertNull(config.clusterInfo()); } @@ -73,7 +73,7 @@ void loadReadsClusterValuesWhenConfigured() { assertNotNull(config.clusterInfo()); assertEquals(2, config.clusterInfo().replicationFactor()); assertEquals(3, config.clusterInfo().nodes().size()); - assertEquals("node-a", config.clusterInfo().nodes().getFirst().id()); + assertEquals("node-a", config.clusterInfo().nodes().getFirst().getId()); } @Test