TerrapinServiceImpl implements TerrapinService.ServiceIface { @Override public Future<TerrapinSingleResponse> get(final TerrapinGetRequest request) { final long startTimeMillis = System.currentTimeMillis(); if (request.getClusterList().isEmpty()) { return Future.exception(new TerrapinGetException("Cluster list is empty", TerrapinGetErrorCode.INVALID_REQUEST)); } ReplicatedTerrapinClient terrapinClient = getReplicatedTerrapinClient(request.getClusterList()); if (terrapinClient == null) { return Future.exception(new TerrapinGetException( "Clusters [" + Joiner.on(", ").join(request.getClusterList()) + "] not found.", TerrapinGetErrorCode.CLUSTER_NOT_FOUND)); } RequestOptions options; if (request.isSetOptions()) { options = request.getOptions(); } else { options = new RequestOptions(); } try { return terrapinClient.getMany(request.getFileSet(), Sets.newHashSet(ByteBuffer.wrap(request.getKey())), options).map( new ExceptionalFunction<TerrapinResponse, TerrapinSingleResponse>() { @Override public TerrapinSingleResponse applyE(TerrapinResponse response) throws TerrapinGetException { ByteBuffer keyBuf = ByteBuffer.wrap(request.getKey()); if (response.getResponseMap().containsKey(keyBuf)) { TerrapinSingleResponse returnResponse = response.getResponseMap().get(keyBuf); if (returnResponse.isSetErrorCode()) { throw new TerrapinGetException("Read failed.", returnResponse.getErrorCode()); } else { Stats.addMetric(request.getFileSet() + "-value-size", returnResponse.getValue().length); Stats.addMetric("value-size", returnResponse.getValue().length); return returnResponse; } } else { return new TerrapinSingleResponse(); } } }).rescue(new Function<Throwable, Future<TerrapinSingleResponse>>() { @Override public Future<TerrapinSingleResponse> apply(Throwable t) { return getExceptionFuture(t); } }).ensure(new Function0<BoxedUnit>() { @Override public BoxedUnit apply() { int timeMillis = (int)(System.currentTimeMillis() - startTimeMillis); Stats.addMetric(request.getFileSet() + "-lookup-latency-ms", timeMillis); Stats.addMetric("lookup-latency-ms", timeMillis); return BoxedUnit.UNIT; } }); } catch (Exception e) { return getExceptionFuture(e); } } TerrapinServiceImpl(PropertiesConfiguration configuration,
                             List<String> clusterList);   TerrapinServiceImpl(Map<String, TerrapinClient> clusterClientMap); @Override Future<TerrapinSingleResponse> get(final TerrapinGetRequest request); @Override Future<TerrapinResponse> multiGet(final TerrapinMultiGetRequest request);  }
@Test public void testGetSuccess() { ByteBuffer key = ByteBuffer.wrap(KEY); TerrapinGetRequest request = prepareGetRequest(); TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of( key, new TerrapinSingleResponse().setValue(ByteBuffer.wrap("value".getBytes())))); when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.value(response)); TerrapinSingleResponse singleResponse = serviceIface.get(request).get(); assertFalse(singleResponse.isSetErrorCode()); assertEquals("value", new String(singleResponse.getValue())); }
@Test public void testGetMultiClusters() { ByteBuffer key = ByteBuffer.wrap(KEY); TerrapinGetRequest request = prepareGetRequest(); RequestOptions options = new RequestOptions(); options.setSelectionPolicy(SelectionPolicy.PRIMARY_FIRST); request.setOptions(options); request.setClusterList(ImmutableList.of(CLUSTER1, CLUSTER2)); TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of( key, new TerrapinSingleResponse().setValue(ByteBuffer.wrap("value".getBytes())))); when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.<TerrapinResponse>exception(new IOException())); when(mockClient2.getManyNoRetries(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.value(response)); TerrapinSingleResponse singleResponse = serviceIface.get(request).get(); assertFalse(singleResponse.isSetErrorCode()); assertEquals("value", new String(singleResponse.getValue())); }
@Test public void testGetEmptyClusters() { ByteBuffer key = ByteBuffer.wrap(KEY); TerrapinGetRequest request = prepareGetRequest(); RequestOptions options = new RequestOptions(); options.setSelectionPolicy(SelectionPolicy.PRIMARY_FIRST); request.setOptions(options); request.setClusterList(ImmutableList.copyOf(new String[]{})); TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of( key, new TerrapinSingleResponse().setValue(ByteBuffer.wrap("value".getBytes())))); when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.value(response)); when(mockClient2.getManyNoRetries(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.value(response)); Try<TerrapinSingleResponse> singleResponseTry = serviceIface.get(request).get(Duration.forever()); assertTrue(singleResponseTry.isThrow()); assertEquals(TerrapinGetErrorCode.INVALID_REQUEST, ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode()); }
@Test public void testGetNotFound() { ByteBuffer key = ByteBuffer.wrap(KEY); TerrapinGetRequest request = prepareGetRequest(); TerrapinResponse response = new TerrapinResponse().setResponseMap((Map)Maps.newHashMap()); when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.value(response)); TerrapinSingleResponse singleResponse = serviceIface.get(request).get(); assertFalse(singleResponse.isSetErrorCode()); assertFalse(singleResponse.isSetValue()); }
@Test public void testGetClusterNotFound() { TerrapinGetRequest request = prepareGetRequest().setClusterList(ImmutableList.of( "random-cluster")); Try<TerrapinSingleResponse> singleResponseTry = serviceIface.get(request).get(Duration.forever()); assertTrue(singleResponseTry.isThrow()); assertEquals(TerrapinGetErrorCode.CLUSTER_NOT_FOUND, ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode()); }
@Test public void testGetError() { ByteBuffer key = ByteBuffer.wrap(KEY); TerrapinGetRequest request = prepareGetRequest(); TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of( key, new TerrapinSingleResponse().setErrorCode(TerrapinGetErrorCode.OTHER))); when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.value(response)); Try<TerrapinSingleResponse> singleResponseTry = serviceIface.get(request).get( Duration.forever()); assertTrue(singleResponseTry.isThrow()); assertEquals(TerrapinGetErrorCode.OTHER, ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode()); when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn( Future.<TerrapinResponse>exception(new TerrapinGetException("Failed.", TerrapinGetErrorCode.FILE_SET_NOT_FOUND))); singleResponseTry = serviceIface.get(request).get(Duration.forever()); assertTrue(singleResponseTry.isThrow()); assertEquals(TerrapinGetErrorCode.FILE_SET_NOT_FOUND, ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode()); when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenThrow( new RuntimeException(new NullPointerException())); singleResponseTry = serviceIface.get(request).get(Duration.forever()); assertTrue(singleResponseTry.isThrow()); assertEquals(TerrapinGetErrorCode.OTHER, ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode()); }