luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fixing fanout cross consumer issue

Browse source
This commit is contained in:
Ashwin Giridharan 2020-02-20 15:48:56 -08:00
parent 097559eca2
commit fedd02c2a5
1 changed files with 5 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRetrievalFactory.java
View file

@ -25,6 +25,8 @@ import software.amazon.kinesis.retrieval.GetRecordsRetrievalStrategy;
import software.amazon.kinesis.retrieval.RecordsPublisher; import software.amazon.kinesis.retrieval.RecordsPublisher;
import software.amazon.kinesis.retrieval.RetrievalFactory; import software.amazon.kinesis.retrieval.RetrievalFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Function; import java.util.function.Function;
@RequiredArgsConstructor @RequiredArgsConstructor
@ -34,6 +36,7 @@ public class FanOutRetrievalFactory implements RetrievalFactory {
private final KinesisAsyncClient kinesisClient; private final KinesisAsyncClient kinesisClient;
private final String defaultStreamName; private final String defaultStreamName;
private final Function<String, String> consumerArnProvider; private final Function<String, String> consumerArnProvider;
private Map<String,String> streamToconsumerArnMap = new HashMap<>();
@Override @Override
public GetRecordsRetrievalStrategy createGetRecordsRetrievalStrategy(final ShardInfo shardInfo, public GetRecordsRetrievalStrategy createGetRecordsRetrievalStrategy(final ShardInfo shardInfo,
@ -45,6 +48,7 @@ public class FanOutRetrievalFactory implements RetrievalFactory {
public RecordsPublisher createGetRecordsCache(@NonNull final ShardInfo shardInfo, public RecordsPublisher createGetRecordsCache(@NonNull final ShardInfo shardInfo,
final MetricsFactory metricsFactory) { final MetricsFactory metricsFactory) {
final String streamName = shardInfo.streamName().orElse(defaultStreamName); final String streamName = shardInfo.streamName().orElse(defaultStreamName);
return new FanOutRecordsPublisher(kinesisClient, shardInfo.shardId(), consumerArnProvider.apply(streamName)); return new FanOutRecordsPublisher(kinesisClient, shardInfo.shardId(),
streamToconsumerArnMap.computeIfAbsent(streamName, consumerArnProvider::apply));
} }
} }