Internally construct and use stream ARNs for all streams in multi-stream mode #1318
Conversation
There was a problem hiding this comment.
Generally lgtm, let's double check with @akidambisrinivasan though
| * @return The {@link Arn} of the Kinesis stream. | ||
| */ | ||
| public static Arn constructStreamArn( | ||
| @NonNull final Region region, final long accountId, @NonNull final String streamName) { |
There was a problem hiding this comment.
Should the accountId just be a String value instead? StreamIdentifier uses accountId as a string. Would avoid any extra parsing we have to do
There was a problem hiding this comment.
Using a long for the accountId can lead to truncation issues for accounts with leading zeros,
so this is actually a bug I think.
Thanks for the callout!
| final Optional<Arn> streamArnOptional = streamConfig.streamIdentifier().streamArnOptional(); | ||
| if (streamArnOptional.isPresent()) { | ||
| Validate.isTrue(kinesisRegion.id().equals(streamArnOptional.get().region().get()), | ||
| "The provided streamARN " + streamArnOptional.get() | ||
| + " does not match the Kinesis client's configured region - " + kinesisRegion); | ||
| return super.put(streamIdentifier, streamConfig); | ||
| } | ||
|
|
||
| if (isMultiStreamMode) { | ||
| return super.put(streamIdentifier, withStreamArn(streamConfig, kinesisRegion)); | ||
| } | ||
|
|
||
| return super.put(streamIdentifier, streamConfig); |
There was a problem hiding this comment.
isPresent() followed by get is an anti-pattern.
Instead use map or ifPresent as appropriate. In this case, something like
| final Optional<Arn> streamArnOptional = streamConfig.streamIdentifier().streamArnOptional(); | |
| if (streamArnOptional.isPresent()) { | |
| Validate.isTrue(kinesisRegion.id().equals(streamArnOptional.get().region().get()), | |
| "The provided streamARN " + streamArnOptional.get() | |
| + " does not match the Kinesis client's configured region - " + kinesisRegion); | |
| return super.put(streamIdentifier, streamConfig); | |
| } | |
| if (isMultiStreamMode) { | |
| return super.put(streamIdentifier, withStreamArn(streamConfig, kinesisRegion)); | |
| } | |
| return super.put(streamIdentifier, streamConfig); | |
| StreamConfig cfg = streamConfig.streamIdentifier().streamArnOptional().map(streamArn -> { | |
| Validate.isTrue(kinesisRegion.id().equals(streamArn.region().get()), | |
| "The provided streamARN " + streamArn + | |
| + " does not match the Kinesis client's configured region - " + kinesisRegion); | |
| return streamConfig; | |
| }).orElse(isMultiStreamMode ? withStreamArn(streamConfig, kinesisRegion):streamConfig); | |
| return super.put(streamIdentifier, cfg); | |
furq-aws
force-pushed
the
multistream-arn-construction
branch
from
c582565 to
6846fe7
Compare
|
|
||
| @KinesisClientInternalApi | ||
| public final class ArnUtil { | ||
| private static final String SERVICE_NAME = "kinesis"; |
There was a problem hiding this comment.
Can you use something already existing e.g. KinesisAsyncClient.SERVICE_NAME ?
There was a problem hiding this comment.
Updated, thank you for pointing this out!
| * Only to be used in multi-stream mode. | ||
| * | ||
| * @param streamConfig The {@link StreamConfig} object to return a copy of. | ||
| * @param kinesisRegion The {@link Region} the stream exists in to be used for constructing the {@link Arn}. |
There was a problem hiding this comment.
nit: missing comma - exists in, to be used for
| scheduler = new Scheduler(checkpointConfig, coordinatorConfig, leaseManagementConfig, lifecycleConfig, | ||
| metricsConfig, processorConfig, retrievalConfig); | ||
|
|
||
| final Scheduler schedulerSpy = spy(scheduler); |
There was a problem hiding this comment.
Why do you need a spy? can you not invoke runProcessLoop on scheduler?
There was a problem hiding this comment.
Good callout! Spy is not needed here; must've copied from a previous unit test and forgotten to clean it up.
furq-aws
dismissed stale reviews from akidambisrinivasan and lucienlu-aws
via
6b7e8e1
furq-aws
force-pushed
the
multistream-arn-construction
branch
from
6846fe7 to
6b7e8e1
Compare
|
|
||
| return super.put(streamIdentifier, streamConfig.streamIdentifier().streamArnOptional() | ||
| .map(streamArn -> { | ||
| Validate.isTrue(kinesisRegion.id().equals(streamArn.region().get()), |
There was a problem hiding this comment.
Just realized that this is invoked in checkAndSyncStreamShardsAndLeases when new streams are discovered, when a shard sync task is generated and updates the currentStreamConfigMap. In that case if the validation fails, it can short-circuit the processing because the exception is only caught in runProcessLoop. This can prevent other streams from being processed. Ideally if there is an issue with a stream, its reasonable for Leader to log an error and skip the stream and maybe even emit a metric. But not let processing of other streams be affected. Can you fix it as a follow up ?
Issue #, if available:
N/A.
Description of changes:
Construct stream ARNs (using the Kinesis client's configured region) for all stream configs upon writing to the
currentStreamConfigMapin multi-stream mode.This should ensure that all Kinesis API requests from
KinesisDataFetcherandKinesisShardDetectorare provided with the stream ARN in multi-stream applications.Note: An implication of stream ARNs being used for Kinesis API calls is that the
accountIdprovided through theStreamTrackermust be correct, whereas previously applications may have consumed the streams as long as the providedstreamNameexisted in the account that the Kinesis client was configured for.By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.