@Test public void testRegisterCalledWhenConsumerArnUnset() throws Exception { FanOutConfig config = new TestingConfig(kinesisClient).applicationName(TEST_APPLICATION_NAME) .streamName(TEST_STREAM_NAME); RetrievalFactory retrievalFactory = config.retrievalFactory(); ShardInfo shardInfo = mock(ShardInfo.class); doReturn(Optional.empty()).when(shardInfo).streamIdentifierSerOpt(); retrievalFactory.createGetRecordsCache(shardInfo, streamConfig, mock(MetricsFactory.class)); assertThat(retrievalFactory, not(nullValue())); verify(consumerRegistration).getOrCreateStreamConsumerArn(); }
@Override public RetrievalFactory retrievalFactory() { return new FanOutRetrievalFactory(kinesisClient, streamName, consumerArn, this::getOrCreateConsumerArn); }
FanOutConfig implements RetrievalSpecificConfig { @Override public RetrievalFactory retrievalFactory() { return new FanOutRetrievalFactory(kinesisClient, streamName, consumerArn, this::getOrCreateConsumerArn); } }
FanOutConfig implements RetrievalSpecificConfig { @Override public RetrievalFactory retrievalFactory() { return new FanOutRetrievalFactory(kinesisClient, streamName, consumerArn, this::getOrCreateConsumerArn); }  }
FanOutConfig implements RetrievalSpecificConfig { @Override public RetrievalFactory retrievalFactory() { return new FanOutRetrievalFactory(kinesisClient, streamName, consumerArn, this::getOrCreateConsumerArn); }  @Override RetrievalFactory retrievalFactory(); }
FanOutConfig implements RetrievalSpecificConfig { @Override public RetrievalFactory retrievalFactory() { return new FanOutRetrievalFactory(kinesisClient, streamName, consumerArn, this::getOrCreateConsumerArn); }  @Override RetrievalFactory retrievalFactory();  }