@Test public void testGetRecordsRetry() throws Exception { Properties kinesisConsumerConfig = new Properties(); kinesisConsumerConfig.setProperty(ConsumerConfigConstants.AWS_REGION, "us-east-1"); final GetRecordsResult expectedResult = new GetRecordsResult(); MutableInt retries = new MutableInt(); final Throwable[] retriableExceptions = new Throwable[] { new AmazonKinesisException("mock"), }; AmazonKinesisClient mockClient = mock(AmazonKinesisClient.class); Mockito.when(mockClient.getRecords(any())).thenAnswer(new Answer<GetRecordsResult>() { @Override public GetRecordsResult answer(InvocationOnMock invocation) throws Throwable{ if (retries.intValue() < retriableExceptions.length) { retries.increment(); throw retriableExceptions[retries.intValue() - 1]; } return expectedResult; } }); KinesisProxy kinesisProxy = new KinesisProxy(kinesisConsumerConfig); Whitebox.getField(KinesisProxy.class, "kinesisClient").set(kinesisProxy, mockClient); GetRecordsResult result = kinesisProxy.getRecords("fakeShardIterator", 1); assertEquals(retriableExceptions.length, retries.intValue()); assertEquals(expectedResult, result); }
@Override public GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet) throws InterruptedException { final GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(shardIterator); getRecordsRequest.setLimit(maxRecordsToGet); GetRecordsResult getRecordsResult = null; int retryCount = 0; while (retryCount <= getRecordsMaxRetries && getRecordsResult == null) { try { getRecordsResult = kinesisClient.getRecords(getRecordsRequest); } catch (SdkClientException ex) { if (isRecoverableSdkClientException(ex)) { long backoffMillis = fullJitterBackoff( getRecordsBaseBackoffMillis, getRecordsMaxBackoffMillis, getRecordsExpConstant, retryCount++); LOG.warn("Got recoverable SdkClientException. Backing off for " + backoffMillis + " millis (" + ex.getClass().getName() + ": " + ex.getMessage() + ")"); Thread.sleep(backoffMillis); } else { throw ex; } } } if (getRecordsResult == null) { throw new RuntimeException("Retries exceeded for getRecords operation - all " + getRecordsMaxRetries + " retry attempts failed."); } return getRecordsResult; }
KinesisProxy implements KinesisProxyInterface { @Override public GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet) throws InterruptedException { final GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(shardIterator); getRecordsRequest.setLimit(maxRecordsToGet); GetRecordsResult getRecordsResult = null; int retryCount = 0; while (retryCount <= getRecordsMaxRetries && getRecordsResult == null) { try { getRecordsResult = kinesisClient.getRecords(getRecordsRequest); } catch (SdkClientException ex) { if (isRecoverableSdkClientException(ex)) { long backoffMillis = fullJitterBackoff( getRecordsBaseBackoffMillis, getRecordsMaxBackoffMillis, getRecordsExpConstant, retryCount++); LOG.warn("Got recoverable SdkClientException. Backing off for " + backoffMillis + " millis (" + ex.getClass().getName() + ": " + ex.getMessage() + ")"); Thread.sleep(backoffMillis); } else { throw ex; } } } if (getRecordsResult == null) { throw new RuntimeException("Retries exceeded for getRecords operation - all " + getRecordsMaxRetries + " retry attempts failed."); } return getRecordsResult; } }
KinesisProxy implements KinesisProxyInterface { @Override public GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet) throws InterruptedException { final GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(shardIterator); getRecordsRequest.setLimit(maxRecordsToGet); GetRecordsResult getRecordsResult = null; int retryCount = 0; while (retryCount <= getRecordsMaxRetries && getRecordsResult == null) { try { getRecordsResult = kinesisClient.getRecords(getRecordsRequest); } catch (SdkClientException ex) { if (isRecoverableSdkClientException(ex)) { long backoffMillis = fullJitterBackoff( getRecordsBaseBackoffMillis, getRecordsMaxBackoffMillis, getRecordsExpConstant, retryCount++); LOG.warn("Got recoverable SdkClientException. Backing off for " + backoffMillis + " millis (" + ex.getClass().getName() + ": " + ex.getMessage() + ")"); Thread.sleep(backoffMillis); } else { throw ex; } } } if (getRecordsResult == null) { throw new RuntimeException("Retries exceeded for getRecords operation - all " + getRecordsMaxRetries + " retry attempts failed."); } return getRecordsResult; } protected  KinesisProxy(Properties configProps); }
KinesisProxy implements KinesisProxyInterface { @Override public GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet) throws InterruptedException { final GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(shardIterator); getRecordsRequest.setLimit(maxRecordsToGet); GetRecordsResult getRecordsResult = null; int retryCount = 0; while (retryCount <= getRecordsMaxRetries && getRecordsResult == null) { try { getRecordsResult = kinesisClient.getRecords(getRecordsRequest); } catch (SdkClientException ex) { if (isRecoverableSdkClientException(ex)) { long backoffMillis = fullJitterBackoff( getRecordsBaseBackoffMillis, getRecordsMaxBackoffMillis, getRecordsExpConstant, retryCount++); LOG.warn("Got recoverable SdkClientException. Backing off for " + backoffMillis + " millis (" + ex.getClass().getName() + ": " + ex.getMessage() + ")"); Thread.sleep(backoffMillis); } else { throw ex; } } } if (getRecordsResult == null) { throw new RuntimeException("Retries exceeded for getRecords operation - all " + getRecordsMaxRetries + " retry attempts failed."); } return getRecordsResult; } protected  KinesisProxy(Properties configProps); static KinesisProxyInterface create(Properties configProps); @Override GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet); @Override GetShardListResult getShardList(Map<String, String> streamNamesWithLastSeenShardIds); @Override String getShardIterator(StreamShardHandle shard, String shardIteratorType, @Nullable Object startingMarker); }
KinesisProxy implements KinesisProxyInterface { @Override public GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet) throws InterruptedException { final GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(shardIterator); getRecordsRequest.setLimit(maxRecordsToGet); GetRecordsResult getRecordsResult = null; int retryCount = 0; while (retryCount <= getRecordsMaxRetries && getRecordsResult == null) { try { getRecordsResult = kinesisClient.getRecords(getRecordsRequest); } catch (SdkClientException ex) { if (isRecoverableSdkClientException(ex)) { long backoffMillis = fullJitterBackoff( getRecordsBaseBackoffMillis, getRecordsMaxBackoffMillis, getRecordsExpConstant, retryCount++); LOG.warn("Got recoverable SdkClientException. Backing off for " + backoffMillis + " millis (" + ex.getClass().getName() + ": " + ex.getMessage() + ")"); Thread.sleep(backoffMillis); } else { throw ex; } } } if (getRecordsResult == null) { throw new RuntimeException("Retries exceeded for getRecords operation - all " + getRecordsMaxRetries + " retry attempts failed."); } return getRecordsResult; } protected  KinesisProxy(Properties configProps); static KinesisProxyInterface create(Properties configProps); @Override GetRecordsResult getRecords(String shardIterator, int maxRecordsToGet); @Override GetShardListResult getShardList(Map<String, String> streamNamesWithLastSeenShardIds); @Override String getShardIterator(StreamShardHandle shard, String shardIteratorType, @Nullable Object startingMarker);  }