/**
 * Test the fail fast mechanism when one asks for more replicas than are configured on the bucket, so it will
 * never be possible to successfully observe the state requested.
 *
 * This test can only be run if less than 3 replicas are defined on the bucket.
 */
@Test
public void shouldFailReplicaIfLessReplicaConfigureOnBucket() {
    Assume.assumeTrue(numberOfReplicas < 3);

    try {
        Observe.call(
            cluster(),
            bucket(),
            "someDoc",
            1234,
            false,
            Observe.PersistTo.NONE,
            Observe.ReplicateTo.THREE,
            BestEffortRetryStrategy.INSTANCE
        ).timeout(5, TimeUnit.SECONDS).toBlocking().single();
        fail("Expected a ReplicaNotConfiguredException");
    } catch (ReplicaNotConfiguredException ex) {
        assertEquals(1234, ex.mutationCas());
    }
}
 

/**
 * Test the fail fast mechanism when one asks for more replicas than are configured on the bucket, so it will
 * never be possible to successfully observe the state requested.
 *
 * This test can only be run if less than 3 replicas are defined on the bucket.
 */
@Test
public void shouldFailPersistIfLessReplicaConfigureOnBucket() {
    Assume.assumeTrue(numberOfReplicas < 3);

    try {
        Observe.call(
            cluster(),
            bucket(),
            "someDoc",
            1234,
            false,
            Observe.PersistTo.FOUR,
            Observe.ReplicateTo.NONE,
            BestEffortRetryStrategy.INSTANCE
        ).timeout(5, TimeUnit.SECONDS).toBlocking().single();
        fail("Expected a ReplicaNotConfiguredException");
    } catch (ReplicaNotConfiguredException ex) {
        assertEquals(1234, ex.mutationCas());
    }
}
 

@Test(expected = ReplicaNotConfiguredException.class)
public void shouldFailFastWhenReplicateToGreaterThanBucketReplicas() {
    ClusterFacade cluster = mock(ClusterFacade.class);

    // Setup a mocked config which returns no replica configured
    CouchbaseBucketConfig bucketConfig = mock(CouchbaseBucketConfig.class);
    when(bucketConfig.numberOfReplicas()).thenReturn(0);
    ClusterConfig clusterConfig = mock(ClusterConfig.class);
    when(clusterConfig.bucketConfig("bucket")).thenReturn(bucketConfig);
    GetClusterConfigResponse clusterConfigResponse = new GetClusterConfigResponse(
        clusterConfig, ResponseStatus.SUCCESS
    );
    when(cluster.send(any(GetClusterConfigRequest.class))).thenReturn(
            Observable.just((CouchbaseResponse) clusterConfigResponse)
    );

    Observable<Boolean> result = Observe.call(
            cluster, "bucket", "id", 1234, false, Observe.PersistTo.NONE, Observe.ReplicateTo.ONE,
            BestEffortRetryStrategy.INSTANCE
    );
    result.toBlocking().single();
}
 

@Test(expected = ReplicaNotConfiguredException.class)
public void shouldFailFastWhenPersistToGreaterThanBucketReplicas() {
    ClusterFacade cluster = mock(ClusterFacade.class);

    // Setup a mocked config which returns no replica configured
    CouchbaseBucketConfig bucketConfig = mock(CouchbaseBucketConfig.class);
    when(bucketConfig.numberOfReplicas()).thenReturn(2);
    ClusterConfig clusterConfig = mock(ClusterConfig.class);
    when(clusterConfig.bucketConfig("bucket")).thenReturn(bucketConfig);
    GetClusterConfigResponse clusterConfigResponse = new GetClusterConfigResponse(
            clusterConfig, ResponseStatus.SUCCESS
    );
    when(cluster.send(any(GetClusterConfigRequest.class))).thenReturn(
            Observable.just((CouchbaseResponse) clusterConfigResponse)
    );

    Observable<Boolean> result = Observe.call(
            cluster, "bucket", "id", 1234, false, Observe.PersistTo.FOUR, Observe.ReplicateTo.NONE,
            BestEffortRetryStrategy.INSTANCE
    );
    result.toBlocking().single();
}
 

/**
 * Fail observables because the partitions do not match up.
 *
 * If the replica is not even available in the configuration (identified by a -2 node index),
 * it is clear that this replica is not configured. If a -1 is returned it is configured, but
 * currently not available (not enough nodes in the cluster, for example if a node is seen down,
 * after a failover, or during rebalance. Replica partitions in general take longer to heal than
 * active partitions, since they are sacrificed for application availability.
 *
 * @param nodeId the current node id of the partition
 * @param request the request to error
 * @param name the name of the bucket
 */
private static void errorObservables(int nodeId, BinaryRequest request, String name, CoreEnvironment env,
    RingBuffer<ResponseEvent> responseBuffer) {
    if (nodeId == DefaultCouchbaseBucketConfig.PARTITION_NOT_EXISTENT) {
        if (request instanceof ReplicaGetRequest) {
            request.observable().onError(new ReplicaNotConfiguredException("Replica number "
                    + ((ReplicaGetRequest) request).replica() + " not configured for bucket " + name, null));
            return;
        } else if (request instanceof ObserveRequest) {
            request.observable().onError(new ReplicaNotConfiguredException("Replica number "
                    + ((ObserveRequest) request).replica() + " not configured for bucket " + name, ((ObserveRequest) request).cas()));
            return;
        } else if (request instanceof ObserveSeqnoRequest) {
            request.observable().onError(new ReplicaNotConfiguredException("Replica number "
                    + ((ObserveSeqnoRequest) request).replica() + " not configured for bucket " + name, ((ObserveSeqnoRequest) request).cas()));
            return;
        }

        RetryHelper.retryOrCancel(env, request, responseBuffer);
        return;
    }

    if (nodeId == -1) {
        if (request instanceof ObserveRequest) {
            request.observable().onError(new ReplicaNotAvailableException("Replica number "
                    + ((ObserveRequest) request).replica() + " not available for bucket " + name,
                ((ObserveRequest) request).cas()));
            return;
        } else if (request instanceof ReplicaGetRequest) {
            request.observable().onError(new ReplicaNotAvailableException("Replica number "
                    + ((ReplicaGetRequest) request).replica() + " not available for bucket " + name, null));
            return;
        } else if (request instanceof ObserveSeqnoRequest) {
            request.observable().onError(new ReplicaNotAvailableException("Replica number "
                    + ((ObserveSeqnoRequest) request).replica() + " not available for bucket " + name, ((ObserveSeqnoRequest) request).cas()));
            return;
        }

        RetryHelper.retryOrCancel(env, request, responseBuffer);
        return;
    }

    throw new IllegalStateException("Unknown NodeId: " + nodeId + ", request: " + request);
}