TerrapinServerInternalImpl implements TerrapinServerInternal.ServiceIface { @Override public Future<TerrapinResponse> get(TerrapinInternalGetRequest request) { final TerrapinResponse response = new TerrapinResponse().setResponseMap( (Map)Maps.newHashMap()); List<Future<Map<ByteBuffer, Pair<ByteBuffer, Throwable>>>> readFutures = Lists.newArrayListWithCapacity(request.getKeyListSize()); String resource = null; for (MultiKey multiKey : request.getKeyList()) { Reader r = null; try { String keyResource = multiKey.getResource(); if (resource != null) { if (!resource.equals(keyResource)) { return Future.exception(new TerrapinGetException( "Multiple resources within same get call.", TerrapinGetErrorCode.INVALID_REQUEST)); } } else { resource = keyResource; } r = this.resourcePartitionMap.getReader(keyResource, multiKey.getPartition()); } catch (TerrapinGetException e) { for (ByteBuffer key : multiKey.getKey()) { response.getResponseMap().put(key, new TerrapinSingleResponse().setErrorCode(e.getErrorCode())); } continue; } try { readFutures.add(r.getValues(multiKey.getKey())); } catch (Throwable t) { t.printStackTrace(); for (ByteBuffer key : multiKey.getKey()) { response.getResponseMap().put(key, new TerrapinSingleResponse().setErrorCode(TerrapinGetErrorCode.READ_ERROR)); } } } return Future.collect(readFutures).map(new Function<List<Map<ByteBuffer, Pair<ByteBuffer, Throwable>>>, TerrapinResponse>() { @Override public TerrapinResponse apply(List<Map<ByteBuffer, Pair<ByteBuffer, Throwable>>> mapList) { for (Map<ByteBuffer, Pair<ByteBuffer, Throwable>> map : mapList) { for (Map.Entry<ByteBuffer, Pair<ByteBuffer, Throwable>> entry : map.entrySet()) { if (entry.getValue().getRight() != null) { response.getResponseMap().put(entry.getKey(), new TerrapinSingleResponse().setErrorCode(TerrapinGetErrorCode.READ_ERROR)); } else { response.getResponseMap().put(entry.getKey(), new TerrapinSingleResponse().setValue(entry.getValue().getLeft())); } } } return response; } }); } TerrapinServerInternalImpl(PropertiesConfiguration configuration,
                                    ResourcePartitionMap resourcePartitionMap); @Override Future<TerrapinResponse> get(TerrapinInternalGetRequest request);  }
@Test public void testGetErrorMultipleResources() throws Exception { TerrapinInternalGetRequest request = new TerrapinInternalGetRequest(); MultiKey multiKey1 = new MultiKey().setResource("resource1").setPartition("1"); multiKey1.addToKey(ByteBuffer.wrap("k1".getBytes())); MultiKey multiKey2 = new MultiKey().setResource("resource2").setPartition("1"); multiKey2.addToKey(ByteBuffer.wrap("k2".getBytes())); request.addToKeyList(multiKey1); request.addToKeyList(multiKey2); Reader mockReader = mock(Reader.class); when(mockResourcePartitionMap.getReader(eq("resource1"), eq("1"))).thenReturn(mockReader); Try<TerrapinResponse> responseTry = serverImpl.get(request).get(Duration.forever()); TerrapinGetException e = (TerrapinGetException)((Throw)responseTry).e(); assertEquals(TerrapinGetErrorCode.INVALID_REQUEST, e.getErrorCode()); }
@Test public void testGetPartialSuccessReaderUnavailable() throws Throwable { Reader mockReader = mock(Reader.class); when(mockResourcePartitionMap.getReader(eq(RESOURCE), eq("1"))).thenThrow( new TerrapinGetException("", TerrapinGetErrorCode.NOT_SERVING_PARTITION)); when(mockResourcePartitionMap.getReader(eq(RESOURCE), eq("2"))).thenReturn(mockReader); when(mockReader.getValues(eq(KEYS_PART_2))).thenAnswer(GET_VALUES_ANSWER); TerrapinResponse response = serverImpl.get(prepareRequest()).get(); Map<ByteBuffer, TerrapinSingleResponse> responseMap = response.getResponseMap(); assertEquals(9, responseMap.size()); for (ByteBuffer key : KEYS_PART_1) { assertEquals(TerrapinGetErrorCode.NOT_SERVING_PARTITION, responseMap.get(key).getErrorCode()); } checkLookup(KEYS_PART_2, responseMap); }
@Test public void testGetPartialSuccess() throws Throwable { Reader mockReader1 = mock(Reader.class); Reader mockReader2 = mock(Reader.class); when(mockResourcePartitionMap.getReader(eq(RESOURCE), eq("1"))).thenReturn(mockReader1); when(mockResourcePartitionMap.getReader(eq(RESOURCE), eq("2"))).thenReturn(mockReader2); when(mockReader1.getValues(eq(KEYS_PART_1))).thenAnswer(GET_VALUES_ANSWER); when(mockReader2.getValues(eq(KEYS_PART_2))).thenAnswer(GET_VALUES_ANSWER); TerrapinResponse response = serverImpl.get(prepareRequest()).get(); Map<ByteBuffer, TerrapinSingleResponse> responseMap = response.getResponseMap(); assertEquals(8, responseMap.size()); checkLookup(KEYS_PART_1, responseMap); checkLookup(KEYS_PART_2, responseMap); }