-
Notifications
You must be signed in to change notification settings - Fork 467
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Experimental support for sequence number validation in the publisher (#…
…401) * Moved sequence number validation to an experimental feature Moved the sequence number validation to become an experimental feature that can be removed in the future. Added an annotation for experimental features. * Delete merge conflict again? * Add some reminder that this stuff is experimental * Added a reason field, and some reasons Added a reason value to the annotation, and updated two of the unusual places.
- Loading branch information
1 parent
01f5db8
commit 592499f
Showing
11 changed files
with
357 additions
and
200 deletions.
There are no files selected for viewing
26 changes: 26 additions & 0 deletions
26
...s-client/src/main/java/software/amazon/kinesis/annotations/KinesisClientExperimental.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| /* | ||
| * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * | ||
| * Licensed under the Amazon Software License (the "License"). | ||
| * You may not use this file except in compliance with the License. | ||
| * A copy of the License is located at | ||
| * | ||
| * http://aws.amazon.com/asl/ | ||
| * | ||
| * or in the "license" file accompanying this file. This file is distributed | ||
| * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
| * express or implied. See the License for the specific language governing | ||
| * permissions and limitations under the License. | ||
| */ | ||
| package software.amazon.kinesis.annotations; | ||
|
|
||
| import java.lang.annotation.Retention; | ||
| import java.lang.annotation.RetentionPolicy; | ||
|
|
||
| /** | ||
| * Anything marked as experimental may be removed at any time without warning. | ||
| */ | ||
| @Retention(RetentionPolicy.CLASS) | ||
| public @interface KinesisClientExperimental { | ||
| String reason() default ""; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
38 changes: 38 additions & 0 deletions
38
.../java/software/amazon/kinesis/retrieval/fanout/experimental/ExperimentalFanOutConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| /* | ||
| * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * | ||
| * Licensed under the Amazon Software License (the "License"). | ||
| * You may not use this file except in compliance with the License. | ||
| * A copy of the License is located at | ||
| * | ||
| * http://aws.amazon.com/asl/ | ||
| * | ||
| * or in the "license" file accompanying this file. This file is distributed | ||
| * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
| * express or implied. See the License for the specific language governing | ||
| * permissions and limitations under the License. | ||
| */ | ||
| package software.amazon.kinesis.retrieval.fanout.experimental; | ||
|
|
||
| import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; | ||
| import software.amazon.kinesis.annotations.KinesisClientExperimental; | ||
| import software.amazon.kinesis.retrieval.RetrievalFactory; | ||
| import software.amazon.kinesis.retrieval.fanout.FanOutConfig; | ||
|
|
||
| /** | ||
| * Enables validation of sequence number for every received record. | ||
| * | ||
| * <h2><strong>This is an experimental class and may be removed at any time</strong></h2> | ||
| */ | ||
| @KinesisClientExperimental | ||
| public class ExperimentalFanOutConfig extends FanOutConfig { | ||
|
|
||
| public ExperimentalFanOutConfig(KinesisAsyncClient kinesisClient) { | ||
| super(kinesisClient); | ||
| } | ||
|
|
||
| @Override | ||
| public RetrievalFactory retrievalFactory() { | ||
| return new ExperimentalFanOutRetrievalFactory(kinesisClient(), getOrCreateConsumerArn()); | ||
| } | ||
| } |
76 changes: 76 additions & 0 deletions
76
...ware/amazon/kinesis/retrieval/fanout/experimental/ExperimentalFanOutRecordsPublisher.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /* | ||
| * Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * | ||
| * Licensed under the Amazon Software License (the "License"). | ||
| * You may not use this file except in compliance with the License. | ||
| * A copy of the License is located at | ||
| * | ||
| * http://aws.amazon.com/asl/ | ||
| * | ||
| * or in the "license" file accompanying this file. This file is distributed | ||
| * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
| * express or implied. See the License for the specific language governing | ||
| * permissions and limitations under the License. | ||
| */ | ||
| package software.amazon.kinesis.retrieval.fanout.experimental; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.function.Function; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.apache.commons.lang3.StringUtils; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; | ||
| import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEvent; | ||
| import software.amazon.kinesis.annotations.KinesisClientExperimental; | ||
| import software.amazon.kinesis.checkpoint.SequenceNumberValidator; | ||
| import software.amazon.kinesis.retrieval.fanout.FanOutRecordsPublisher; | ||
|
|
||
| /** | ||
| * A variation of {@link FanOutRecordsPublisher} that provides validation of every record received by the publisher. | ||
| * | ||
| * <h2><strong>This is an experimental class and may be removed at any time</strong></h2> | ||
| */ | ||
| @Slf4j | ||
| @KinesisClientExperimental | ||
| public class ExperimentalFanOutRecordsPublisher extends FanOutRecordsPublisher { | ||
|
|
||
| private final SequenceNumberValidator sequenceNumberValidator = new SequenceNumberValidator(); | ||
|
|
||
| /** | ||
| * Creates a new FanOutRecordsPublisher. | ||
| * | ||
| * @param kinesis | ||
| * the kinesis client to use for requests | ||
| * @param shardId | ||
| * the shardId to retrieve records for | ||
| * @param consumerArn | ||
| */ | ||
| public ExperimentalFanOutRecordsPublisher(KinesisAsyncClient kinesis, String shardId, String consumerArn) { | ||
| super(kinesis, shardId, consumerArn); | ||
| } | ||
|
|
||
| @Override | ||
| protected void validateRecords(String shardId, SubscribeToShardEvent event) { | ||
| Map<String, Integer> mismatchedRecords = recordsNotForShard(shardId, event); | ||
| if (mismatchedRecords.size() > 0) { | ||
| String mismatchReport = mismatchedRecords.entrySet().stream() | ||
| .map(e -> String.format("(%s -> %d)", e.getKey(), e.getValue())).collect(Collectors.joining(", ")); | ||
| throw new IllegalArgumentException("Received records destined for different shards: " + mismatchReport); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| private Map<String, Integer> recordsNotForShard(String shardId, SubscribeToShardEvent event) { | ||
| return event.records().stream().map(r -> { | ||
| Optional<String> res = sequenceNumberValidator.shardIdFor(r.sequenceNumber()); | ||
| if (!res.isPresent()) { | ||
| throw new IllegalArgumentException("Unable to validate sequence number of " + r.sequenceNumber()); | ||
| } | ||
| return res.get(); | ||
| }).filter(s -> !StringUtils.equalsIgnoreCase(s, shardId)).collect(Collectors.groupingBy(Function.identity())) | ||
| .entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().size())); | ||
| } | ||
| } |
Oops, something went wrong.