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

Provide streamArn in getRecords request (#1219)

Browse source
This commit is contained in:
furq-aws 2023-10-26 16:49:07 -07:00 committed by GitHub
parent 118783b18b
commit f90b1b1c05
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisShardDetector.java
View file

@ -307,23 +307,26 @@ public class KinesisShardDetector implements ShardDetector {
}
@Override
public List<ChildShard> getChildShards(final String shardId) throws InterruptedException, ExecutionException, TimeoutException {
final GetShardIteratorRequest.Builder requestBuilder = KinesisRequestsBuilder.getShardIteratorRequestBuilder()
.streamName(streamIdentifier.streamName())
.shardIteratorType(ShardIteratorType.LATEST)
.shardId(shardId);
streamIdentifier.streamArnOptional().ifPresent(arn -> requestBuilder.streamARN(arn.toString()));
final GetShardIteratorRequest getShardIteratorRequest = requestBuilder.build();
public List<ChildShard> getChildShards(final String shardId)
throws InterruptedException, ExecutionException, TimeoutException {
final GetShardIteratorRequest.Builder getShardIteratorRequestBuilder =
KinesisRequestsBuilder.getShardIteratorRequestBuilder()
.streamName(streamIdentifier.streamName())
.shardIteratorType(ShardIteratorType.LATEST)
.shardId(shardId);
streamIdentifier.streamArnOptional().ifPresent(arn -> getShardIteratorRequestBuilder.streamARN(arn.toString()));
final GetShardIteratorResponse getShardIteratorResponse =
FutureUtils.resolveOrCancelFuture(kinesisClient.getShardIterator(getShardIteratorRequest), kinesisRequestTimeout);
final GetShardIteratorResponse getShardIteratorResponse = FutureUtils.resolveOrCancelFuture(
kinesisClient.getShardIterator(getShardIteratorRequestBuilder.build()),
kinesisRequestTimeout);
final GetRecordsRequest getRecordsRequest = KinesisRequestsBuilder.getRecordsRequestBuilder()
.shardIterator(getShardIteratorResponse.shardIterator())
.build();
final GetRecordsRequest.Builder getRecordsRequestBuilder = KinesisRequestsBuilder.getRecordsRequestBuilder()
.shardIterator(getShardIteratorResponse.shardIterator());
streamIdentifier.streamArnOptional().ifPresent(arn -> getRecordsRequestBuilder.streamARN(arn.toString()));
final GetRecordsResponse getRecordsResponse =
FutureUtils.resolveOrCancelFuture(kinesisClient.getRecords(getRecordsRequest), kinesisRequestTimeout);
final GetRecordsResponse getRecordsResponse = FutureUtils.resolveOrCancelFuture(
kinesisClient.getRecords(getRecordsRequestBuilder.build()),
kinesisRequestTimeout);
return getRecordsResponse.childShards();
}