@Test void shouldDistributeFairlyUnderMDS() { Locator locator = new RoundRobinLocator(ServiceType.QUERY, 0); QueryRequest request = mock(QueryRequest.class); ClusterConfig configMock = mock(ClusterConfig.class); when(configMock.hasClusterOrBucketConfig()).thenReturn(true); Node node1Mock = mock(Node.class); when(node1Mock.identifier()).thenReturn(new NodeIdentifier("192.168.56.101", 8091)); when(node1Mock.serviceEnabled(ServiceType.QUERY)).thenReturn(false); Node node2Mock = mock(Node.class); when(node2Mock.identifier()).thenReturn(new NodeIdentifier("192.168.56.102", 8091)); when(node2Mock.serviceEnabled(ServiceType.QUERY)).thenReturn(false); Node node3Mock = mock(Node.class); when(node3Mock.identifier()).thenReturn(new NodeIdentifier("192.168.56.103", 8091)); when(node3Mock.serviceEnabled(ServiceType.QUERY)).thenReturn(true); Node node4Mock = mock(Node.class); when(node4Mock.identifier()).thenReturn(new NodeIdentifier("192.168.56.104", 8091)); when(node4Mock.serviceEnabled(ServiceType.QUERY)).thenReturn(true); List<Node> nodes = new ArrayList<>(Arrays.asList(node1Mock, node2Mock, node3Mock, node4Mock)); locator.dispatch(request, nodes, configMock, null); verify(node1Mock, never()).send(request); verify(node2Mock, never()).send(request); verify(node3Mock, times(1)).send(request); verify(node4Mock, never()).send(request); locator.dispatch(request, nodes, configMock, null); verify(node1Mock, never()).send(request); verify(node2Mock, never()).send(request); verify(node3Mock, times(1)).send(request); verify(node4Mock, times(1)).send(request); locator.dispatch(request, nodes, configMock, null); verify(node1Mock, never()).send(request); verify(node2Mock, never()).send(request); verify(node3Mock, times(2)).send(request); verify(node4Mock, times(1)).send(request); locator.dispatch(request, nodes, configMock, null); verify(node1Mock, never()).send(request); verify(node2Mock, never()).send(request); verify(node3Mock, times(2)).send(request); verify(node4Mock, times(2)).send(request); }
@Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (!checkServiceNotAvailable(request, config)) { return; } boolean isTargeted = request instanceof TargetedRequest; if (!isTargeted && !config.hasClusterOrBucketConfig()) { boolean globalLoadInProgress = ctx.core().configurationProvider().globalConfigLoadInProgress(); boolean bucketLoadInProgress =ctx.core().configurationProvider().bucketConfigLoadInProgress(); boolean loadInProgress = globalLoadInProgress || bucketLoadInProgress; if (loadInProgress) { RetryOrchestrator.maybeRetry( ctx, request, bucketLoadInProgress ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.GLOBAL_CONFIG_LOAD_IN_PROGRESS ); } else { request.fail(FeatureNotAvailableException.clusterLevelQuery(serviceType)); } return; } List<Node> filteredNodes = filterNodes(nodes, request, config); if (filteredNodes.isEmpty()) { if (serviceShowsUpInConfig(config)) { RetryOrchestrator.maybeRetry(ctx, request, RetryReason.NODE_NOT_AVAILABLE); } else { request.fail(new ServiceNotAvailableException( "The " + request.serviceType().ident() + " service is not available in the cluster.", new GenericRequestErrorContext(request) )); } return; } if (isTargeted) { dispatchTargeted(request, filteredNodes, ctx); } else { dispatchUntargeted(request, filteredNodes, ctx); } }
RoundRobinLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (!checkServiceNotAvailable(request, config)) { return; } boolean isTargeted = request instanceof TargetedRequest; if (!isTargeted && !config.hasClusterOrBucketConfig()) { boolean globalLoadInProgress = ctx.core().configurationProvider().globalConfigLoadInProgress(); boolean bucketLoadInProgress =ctx.core().configurationProvider().bucketConfigLoadInProgress(); boolean loadInProgress = globalLoadInProgress || bucketLoadInProgress; if (loadInProgress) { RetryOrchestrator.maybeRetry( ctx, request, bucketLoadInProgress ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.GLOBAL_CONFIG_LOAD_IN_PROGRESS ); } else { request.fail(FeatureNotAvailableException.clusterLevelQuery(serviceType)); } return; } List<Node> filteredNodes = filterNodes(nodes, request, config); if (filteredNodes.isEmpty()) { if (serviceShowsUpInConfig(config)) { RetryOrchestrator.maybeRetry(ctx, request, RetryReason.NODE_NOT_AVAILABLE); } else { request.fail(new ServiceNotAvailableException( "The " + request.serviceType().ident() + " service is not available in the cluster.", new GenericRequestErrorContext(request) )); } return; } if (isTargeted) { dispatchTargeted(request, filteredNodes, ctx); } else { dispatchUntargeted(request, filteredNodes, ctx); } } }
RoundRobinLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (!checkServiceNotAvailable(request, config)) { return; } boolean isTargeted = request instanceof TargetedRequest; if (!isTargeted && !config.hasClusterOrBucketConfig()) { boolean globalLoadInProgress = ctx.core().configurationProvider().globalConfigLoadInProgress(); boolean bucketLoadInProgress =ctx.core().configurationProvider().bucketConfigLoadInProgress(); boolean loadInProgress = globalLoadInProgress || bucketLoadInProgress; if (loadInProgress) { RetryOrchestrator.maybeRetry( ctx, request, bucketLoadInProgress ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.GLOBAL_CONFIG_LOAD_IN_PROGRESS ); } else { request.fail(FeatureNotAvailableException.clusterLevelQuery(serviceType)); } return; } List<Node> filteredNodes = filterNodes(nodes, request, config); if (filteredNodes.isEmpty()) { if (serviceShowsUpInConfig(config)) { RetryOrchestrator.maybeRetry(ctx, request, RetryReason.NODE_NOT_AVAILABLE); } else { request.fail(new ServiceNotAvailableException( "The " + request.serviceType().ident() + " service is not available in the cluster.", new GenericRequestErrorContext(request) )); } return; } if (isTargeted) { dispatchTargeted(request, filteredNodes, ctx); } else { dispatchUntargeted(request, filteredNodes, ctx); } } RoundRobinLocator(final ServiceType serviceType);   RoundRobinLocator(final ServiceType serviceType, final long initialValue); }
RoundRobinLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (!checkServiceNotAvailable(request, config)) { return; } boolean isTargeted = request instanceof TargetedRequest; if (!isTargeted && !config.hasClusterOrBucketConfig()) { boolean globalLoadInProgress = ctx.core().configurationProvider().globalConfigLoadInProgress(); boolean bucketLoadInProgress =ctx.core().configurationProvider().bucketConfigLoadInProgress(); boolean loadInProgress = globalLoadInProgress || bucketLoadInProgress; if (loadInProgress) { RetryOrchestrator.maybeRetry( ctx, request, bucketLoadInProgress ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.GLOBAL_CONFIG_LOAD_IN_PROGRESS ); } else { request.fail(FeatureNotAvailableException.clusterLevelQuery(serviceType)); } return; } List<Node> filteredNodes = filterNodes(nodes, request, config); if (filteredNodes.isEmpty()) { if (serviceShowsUpInConfig(config)) { RetryOrchestrator.maybeRetry(ctx, request, RetryReason.NODE_NOT_AVAILABLE); } else { request.fail(new ServiceNotAvailableException( "The " + request.serviceType().ident() + " service is not available in the cluster.", new GenericRequestErrorContext(request) )); } return; } if (isTargeted) { dispatchTargeted(request, filteredNodes, ctx); } else { dispatchUntargeted(request, filteredNodes, ctx); } } RoundRobinLocator(final ServiceType serviceType);   RoundRobinLocator(final ServiceType serviceType, final long initialValue); @Override void dispatch(final Request<? extends Response> request, final List<Node> nodes,
                       final ClusterConfig config, final CoreContext ctx); }
RoundRobinLocator implements Locator { @Override public void dispatch(final Request<? extends Response> request, final List<Node> nodes, final ClusterConfig config, final CoreContext ctx) { if (!checkServiceNotAvailable(request, config)) { return; } boolean isTargeted = request instanceof TargetedRequest; if (!isTargeted && !config.hasClusterOrBucketConfig()) { boolean globalLoadInProgress = ctx.core().configurationProvider().globalConfigLoadInProgress(); boolean bucketLoadInProgress =ctx.core().configurationProvider().bucketConfigLoadInProgress(); boolean loadInProgress = globalLoadInProgress || bucketLoadInProgress; if (loadInProgress) { RetryOrchestrator.maybeRetry( ctx, request, bucketLoadInProgress ? RetryReason.BUCKET_OPEN_IN_PROGRESS : RetryReason.GLOBAL_CONFIG_LOAD_IN_PROGRESS ); } else { request.fail(FeatureNotAvailableException.clusterLevelQuery(serviceType)); } return; } List<Node> filteredNodes = filterNodes(nodes, request, config); if (filteredNodes.isEmpty()) { if (serviceShowsUpInConfig(config)) { RetryOrchestrator.maybeRetry(ctx, request, RetryReason.NODE_NOT_AVAILABLE); } else { request.fail(new ServiceNotAvailableException( "The " + request.serviceType().ident() + " service is not available in the cluster.", new GenericRequestErrorContext(request) )); } return; } if (isTargeted) { dispatchTargeted(request, filteredNodes, ctx); } else { dispatchUntargeted(request, filteredNodes, ctx); } } RoundRobinLocator(final ServiceType serviceType);   RoundRobinLocator(final ServiceType serviceType, final long initialValue); @Override void dispatch(final Request<? extends Response> request, final List<Node> nodes,
                       final ClusterConfig config, final CoreContext ctx);  }