diff --git a/src/main/java/org/cache/Main.java b/src/main/java/org/cache/Main.java index b3e6728..7df0af4 100644 --- a/src/main/java/org/cache/Main.java +++ b/src/main/java/org/cache/Main.java @@ -103,6 +103,24 @@ public CacheOperations routedCacheService( return new RoutedCacheService<>(cacheService, cacheNode, cacheConfig.clusterInfo(), forwardingClient, keyCodec); } + @Bean + public CacheOperations clusterRoutedCacheService( + CacheService cacheService, + CacheNode cacheNode, + CacheConfig cacheConfig, + ClusterForwardingClient forwardingClient, + KeyCodec keyCodec + ) { + return new RoutedCacheService<>( + cacheService, + cacheNode, + cacheConfig.clusterInfo(), + forwardingClient, + keyCodec, + false + ); + } + @Bean public CommandProcessor commandProcessor(KeyCodec keyCodec, CacheOperations cacheService) { return new CommandProcessor<>(keyCodec, cacheService); @@ -111,7 +129,7 @@ public CommandProcessor commandProcessor(KeyCodec keyCodec, Cach @Bean public CommandProcessor clusterCommandProcessor( KeyCodec keyCodec, - CacheService cacheService + @Qualifier("clusterRoutedCacheService") CacheOperations cacheService ) { return new CommandProcessor<>(keyCodec, cacheService); } diff --git a/src/main/java/org/cache/cluster/routing/RoutedCacheService.java b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java index ce598e6..ea0e24c 100644 --- a/src/main/java/org/cache/cluster/routing/RoutedCacheService.java +++ b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java @@ -19,6 +19,7 @@ public class RoutedCacheService implements CacheOperations { private final ConsistentHashRing hashRing; private final KeyCodec keyCodec; private final CacheResponseParser responseParser; + private final boolean forwardingAllowed; public RoutedCacheService( CacheOperations localService, @@ -27,7 +28,18 @@ public RoutedCacheService( ClusterForwardingClient forwardingClient, KeyCodec keyCodec ) { - this(localService, currentNode, clusterInfo, forwardingClient, keyCodec, new CacheResponseParser()); + this(localService, currentNode, clusterInfo, forwardingClient, keyCodec, true); + } + + public RoutedCacheService( + CacheOperations localService, + CacheNode currentNode, + ClusterInfo clusterInfo, + ClusterForwardingClient forwardingClient, + KeyCodec keyCodec, + boolean forwardingAllowed + ) { + this(localService, currentNode, clusterInfo, forwardingClient, keyCodec, new CacheResponseParser(), forwardingAllowed); } RoutedCacheService( @@ -36,7 +48,8 @@ public RoutedCacheService( ClusterInfo clusterInfo, ClusterForwardingClient forwardingClient, KeyCodec keyCodec, - CacheResponseParser responseParser + CacheResponseParser responseParser, + boolean forwardingAllowed ) { this.localService = localService; this.currentNode = currentNode; @@ -44,17 +57,18 @@ public RoutedCacheService( this.hashRing = clusterInfo == null ? null : new ConsistentHashRing(clusterInfo); this.keyCodec = keyCodec; this.responseParser = responseParser; + this.forwardingAllowed = forwardingAllowed; } @Override public void putString(K key, String value, long ttlMillis) { - CacheNode owner = ownerFor(key); - if (isLocal(owner)) { + Optional remoteOwner = remoteOwnerFor(key); + if (remoteOwner.isEmpty()) { localService.putString(key, value, ttlMillis); return; } - expectOk(forwardingClient.forward(owner, List.of( + expectOk(forwardingClient.forward(remoteOwner.get(), List.of( "PUT", keyCodec.encode(key), value, @@ -64,12 +78,12 @@ public void putString(K key, String value, long ttlMillis) { @Override public Optional getString(K key) { - CacheNode owner = ownerFor(key); - if (isLocal(owner)) { + Optional remoteOwner = remoteOwnerFor(key); + if (remoteOwner.isEmpty()) { return localService.getString(key); } - List response = forwardingClient.forward(owner, List.of("GET", keyCodec.encode(key))); + List response = forwardingClient.forward(remoteOwner.get(), List.of("GET", keyCodec.encode(key))); if (responseParser.isNotFound(response)) { return Optional.empty(); } @@ -82,23 +96,23 @@ public Optional getString(K key) { @Override public void push(K key, String value) { - CacheNode owner = ownerFor(key); - if (isLocal(owner)) { + Optional remoteOwner = remoteOwnerFor(key); + if (remoteOwner.isEmpty()) { localService.push(key, value); return; } - expectOk(forwardingClient.forward(owner, List.of("PUSH", keyCodec.encode(key), value))); + expectOk(forwardingClient.forward(remoteOwner.get(), List.of("PUSH", keyCodec.encode(key), value))); } @Override public Optional> lrange(K key, int from, int to) { - CacheNode owner = ownerFor(key); - if (isLocal(owner)) { + Optional remoteOwner = remoteOwnerFor(key); + if (remoteOwner.isEmpty()) { return localService.lrange(key, from, to); } - List response = forwardingClient.forward(owner, List.of( + List response = forwardingClient.forward(remoteOwner.get(), List.of( "LRANGE", keyCodec.encode(key), Integer.toString(from), @@ -122,13 +136,13 @@ public Optional> lrange(K key, int from, int to) { @Override public void delete(K key) { - CacheNode owner = ownerFor(key); - if (isLocal(owner)) { + Optional remoteOwner = remoteOwnerFor(key); + if (remoteOwner.isEmpty()) { localService.delete(key); return; } - expectOk(forwardingClient.forward(owner, List.of("DELETE", keyCodec.encode(key)))); + expectOk(forwardingClient.forward(remoteOwner.get(), List.of("DELETE", keyCodec.encode(key)))); } @Override @@ -146,12 +160,18 @@ public Snapshot metrics() { return localService.metrics(); } - private CacheNode ownerFor(K key) { - return hashRing == null ? currentNode : hashRing.nodeFor(keyCodec.encode(key)); - } + private Optional remoteOwnerFor(K key) { + CacheNode owner = hashRing == null ? currentNode : hashRing.nodeFor(keyCodec.encode(key)); + if (owner.id().equals(currentNode.id())) { + return Optional.empty(); + } + + if (!forwardingAllowed) { + throw new ClusterForwardingException("Request routed to wrong node. Expected owner: " + owner.id() + + ", current node: " + currentNode.id()); + } - private boolean isLocal(CacheNode owner) { - return owner.id().equals(currentNode.id()); + return Optional.of(owner); } private void expectOk(List response) { diff --git a/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java index a9bf795..68dd4ee 100644 --- a/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java +++ b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java @@ -11,6 +11,7 @@ import java.util.Optional; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; @@ -63,6 +64,26 @@ void getStringForwardsWhenAnotherNodeOwnsKey() { verifyNoInteractions(localService); } + @Test + void getStringRejectsRemoteOwnerWhenForwardingIsDisabled() { + CacheOperations localService = localService(); + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + String key = keyOwnedBy(nodeB); + RoutedCacheService service = new RoutedCacheService<>( + localService, + nodeA, + clusterInfo, + forwardingClient, + keyCodec, + false + ); + + var exception = assertThrows(ClusterForwardingException.class, () -> service.getString(key)); + + assertEquals("Request routed to wrong node. Expected owner: node-b, current node: node-a", exception.getMessage()); + verifyNoInteractions(localService, forwardingClient); + } + @Test void lrangeForwardsAndReturnsListValues() { CacheOperations localService = localService();