@Test @SuppressWarnings("unchecked") void pickFastForwardIfAvailableAndRetry() { Locator locator = new KeyValueLocator(); NodeInfo nodeInfo1 = new NodeInfo("http: Collections.EMPTY_MAP, null); NodeInfo nodeInfo2 = new NodeInfo("http: Collections.EMPTY_MAP, null); Node node1Mock = mock(Node.class); when(node1Mock.identifier()).thenReturn(new NodeIdentifier("192.168.56.101", 8091)); Node node2Mock = mock(Node.class); when(node2Mock.identifier()).thenReturn(new NodeIdentifier("192.168.56.102", 8091)); List<Node> nodes = new ArrayList<>(Arrays.asList(node1Mock, node2Mock)); ClusterConfig configMock = mock(ClusterConfig.class); CouchbaseBucketConfig bucketMock = mock(CouchbaseBucketConfig.class); when(configMock.bucketConfig("bucket")).thenReturn(bucketMock); when(bucketMock.nodes()).thenReturn(Arrays.asList(nodeInfo1, nodeInfo2)); when(bucketMock.numberOfPartitions()).thenReturn(1024); when(bucketMock.nodeAtIndex(0)).thenReturn(nodeInfo1); when(bucketMock.nodeAtIndex(1)).thenReturn(nodeInfo2); when(bucketMock.hasFastForwardMap()).thenReturn(true); when(bucketMock.nodeIndexForActive(656, false)).thenReturn((short) 0); when(bucketMock.nodeIndexForActive(656, true)).thenReturn((short) 1); GetRequest getRequest = mock(GetRequest.class); when(getRequest.bucket()).thenReturn("bucket"); when(getRequest.key()).thenReturn("key".getBytes(UTF_8)); RequestContext requestCtx = mock(RequestContext.class); when(getRequest.context()).thenReturn(requestCtx); when(requestCtx.retryAttempts()).thenReturn(0); locator.dispatch(getRequest, nodes, configMock, null); verify(node1Mock, times(1)).send(getRequest); verify(node2Mock, never()).send(getRequest); when(requestCtx.retryAttempts()).thenReturn(1); locator.dispatch(getRequest, nodes, configMock, null); verify(node1Mock, times(1)).send(getRequest); verify(node2Mock, times(1)).send(getRequest); when(requestCtx.retryAttempts()).thenReturn(2); locator.dispatch(getRequest, nodes, configMock, null); verify(node1Mock, times(2)).send(getRequest); verify(node2Mock, times(1)).send(getRequest); when(requestCtx.retryAttempts()).thenReturn(3); locator.dispatch(getRequest, nodes, configMock, null); verify(node1Mock, times(2)).send(getRequest); verify(node2Mock, times(2)).send(getRequest); when(requestCtx.retryAttempts()).thenReturn(4); locator.dispatch(getRequest, nodes, configMock, null); verify(node1Mock, times(3)).send(getRequest); verify(node2Mock, times(2)).send(getRequest); }
@Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (request instanceof TargetedRequest) { dispatchTargeted((TargetedRequest) request, nodes, ctx); } else { KeyValueRequest r = (KeyValueRequest) request; String bucket = r.bucket(); BucketConfig bucketConfig = config.bucketConfig(bucket); if (bucketConfig == null) { RetryOrchestrator.maybeRetry(ctx, request, ctx.core().configurationProvider().bucketConfigLoadInProgress() ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.BUCKET_NOT_AVAILABLE); return; } if (bucketConfig instanceof CouchbaseBucketConfig) { couchbaseBucket(r, nodes, (CouchbaseBucketConfig) bucketConfig, ctx); } else if (bucketConfig instanceof MemcachedBucketConfig) { memcacheBucket(r, nodes, (MemcachedBucketConfig) bucketConfig, ctx); } else { throw new IllegalStateException("Unsupported Bucket Type: " + bucketConfig + " for request " + request); } } }
KeyValueLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (request instanceof TargetedRequest) { dispatchTargeted((TargetedRequest) request, nodes, ctx); } else { KeyValueRequest r = (KeyValueRequest) request; String bucket = r.bucket(); BucketConfig bucketConfig = config.bucketConfig(bucket); if (bucketConfig == null) { RetryOrchestrator.maybeRetry(ctx, request, ctx.core().configurationProvider().bucketConfigLoadInProgress() ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.BUCKET_NOT_AVAILABLE); return; } if (bucketConfig instanceof CouchbaseBucketConfig) { couchbaseBucket(r, nodes, (CouchbaseBucketConfig) bucketConfig, ctx); } else if (bucketConfig instanceof MemcachedBucketConfig) { memcacheBucket(r, nodes, (MemcachedBucketConfig) bucketConfig, ctx); } else { throw new IllegalStateException("Unsupported Bucket Type: " + bucketConfig + " for request " + request); } } } }
KeyValueLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (request instanceof TargetedRequest) { dispatchTargeted((TargetedRequest) request, nodes, ctx); } else { KeyValueRequest r = (KeyValueRequest) request; String bucket = r.bucket(); BucketConfig bucketConfig = config.bucketConfig(bucket); if (bucketConfig == null) { RetryOrchestrator.maybeRetry(ctx, request, ctx.core().configurationProvider().bucketConfigLoadInProgress() ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.BUCKET_NOT_AVAILABLE); return; } if (bucketConfig instanceof CouchbaseBucketConfig) { couchbaseBucket(r, nodes, (CouchbaseBucketConfig) bucketConfig, ctx); } else if (bucketConfig instanceof MemcachedBucketConfig) { memcacheBucket(r, nodes, (MemcachedBucketConfig) bucketConfig, ctx); } else { throw new IllegalStateException("Unsupported Bucket Type: " + bucketConfig + " for request " + request); } } }  }
KeyValueLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (request instanceof TargetedRequest) { dispatchTargeted((TargetedRequest) request, nodes, ctx); } else { KeyValueRequest r = (KeyValueRequest) request; String bucket = r.bucket(); BucketConfig bucketConfig = config.bucketConfig(bucket); if (bucketConfig == null) { RetryOrchestrator.maybeRetry(ctx, request, ctx.core().configurationProvider().bucketConfigLoadInProgress() ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.BUCKET_NOT_AVAILABLE); return; } if (bucketConfig instanceof CouchbaseBucketConfig) { couchbaseBucket(r, nodes, (CouchbaseBucketConfig) bucketConfig, ctx); } else if (bucketConfig instanceof MemcachedBucketConfig) { memcacheBucket(r, nodes, (MemcachedBucketConfig) bucketConfig, ctx); } else { throw new IllegalStateException("Unsupported Bucket Type: " + bucketConfig + " for request " + request); } } }  @Override void dispatch(final Request<? extends Response> request, final List<Node> nodes,
                       final ClusterConfig config, final CoreContext ctx); static int partitionForKey(final byte[] id, final int numPartitions); }
KeyValueLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (request instanceof TargetedRequest) { dispatchTargeted((TargetedRequest) request, nodes, ctx); } else { KeyValueRequest r = (KeyValueRequest) request; String bucket = r.bucket(); BucketConfig bucketConfig = config.bucketConfig(bucket); if (bucketConfig == null) { RetryOrchestrator.maybeRetry(ctx, request, ctx.core().configurationProvider().bucketConfigLoadInProgress() ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.BUCKET_NOT_AVAILABLE); return; } if (bucketConfig instanceof CouchbaseBucketConfig) { couchbaseBucket(r, nodes, (CouchbaseBucketConfig) bucketConfig, ctx); } else if (bucketConfig instanceof MemcachedBucketConfig) { memcacheBucket(r, nodes, (MemcachedBucketConfig) bucketConfig, ctx); } else { throw new IllegalStateException("Unsupported Bucket Type: " + bucketConfig + " for request " + request); } } }  @Override void dispatch(final Request<? extends Response> request, final List<Node> nodes,
                       final ClusterConfig config, final CoreContext ctx); static int partitionForKey(final byte[] id, final int numPartitions);  }