From d8fd4c2aabaa67c05064588ba5c17cd1970bf198 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 27 Apr 2022 12:54:05 -0400 Subject: [PATCH] GH-2359: Enable other KCL initial offsets Fixes https://github.com/spring-cloud/spring-cloud-stream/issues/2359 The KCL mode does not apply at the moment `AT_TIMESTAMP` and `TRIM_HORIZON` initial offsets Resolves #2365 * Change the `KinesisMessageChannelBinder` binder logic to create an explicit `KinesisClientLibConfiguration` based on the provided properties including custom stream iterator type * Adjust `KinesisTestBinder` to be able to create a KCL/KPL based bindings --- .../kinesis/KinesisMessageChannelBinder.java | 116 +++++++++++++----- .../binder/kinesis/KinesisBinderTests.java | 55 ++++++++- .../binder/kinesis/KinesisTestBinder.java | 21 ++-- 3 files changed, 156 insertions(+), 36 deletions(-) diff --git a/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java b/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java index 6f862b0d2..3c51d0dd3 100644 --- a/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java +++ b/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2020 the original author or authors. + * Copyright 2017-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kinesis; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -26,6 +27,7 @@ import java.util.Objects; import java.util.Set; import java.util.UUID; +import com.amazonaws.ClientConfiguration; import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.services.cloudwatch.AmazonCloudWatch; import com.amazonaws.services.dynamodbv2.AmazonDynamoDB; @@ -36,6 +38,7 @@ import com.amazonaws.services.kinesis.AmazonKinesis; import com.amazonaws.services.kinesis.AmazonKinesisAsync; import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream; import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration; +import com.amazonaws.services.kinesis.clientlibrary.lib.worker.SimpleRecordsFetcherFactory; import com.amazonaws.services.kinesis.model.InvalidArgumentException; import com.amazonaws.services.kinesis.model.Shard; import com.amazonaws.services.kinesis.model.ShardIteratorType; @@ -341,8 +344,6 @@ public class KinesisMessageChannelBinder extends "Ignoring 'shardId' property"); } - String shardIteratorType = kinesisConsumerProperties.getShardIteratorType(); - AmazonKinesis amazonKinesisClient = kinesisConsumerProperties.isDynamoDbStreams() ? this.dynamoDBStreamsAdapter @@ -350,31 +351,13 @@ public class KinesisMessageChannelBinder extends String stream = destination.getName(); - KinesisClientLibConfiguration kinesisClientLibConfiguration = obtainKinesisClientLibConfiguration(stream, group); + KinesisClientLibConfiguration kinesisClientLibConfiguration = + obtainKinesisClientLibConfiguration(properties.getExtension(), stream, group); - KclMessageDrivenChannelAdapter adapter; - - String consumerGroup; - if (kinesisClientLibConfiguration == null) { - adapter = new KclMessageDrivenChannelAdapter(stream, amazonKinesisClient, this.cloudWatchClient, - this.dynamoDBClient, this.awsCredentialsProvider); - boolean anonymous = !StringUtils.hasText(group); - consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group; - adapter.setConsumerGroup(consumerGroup); - if (StringUtils.hasText(shardIteratorType)) { - adapter.setStreamInitialSequence(InitialPositionInStream.valueOf(shardIteratorType)); - } - adapter.setIdleBetweenPolls(kinesisConsumerProperties.getIdleBetweenPolls()); - adapter.setConsumerBackoff(kinesisConsumerProperties.getConsumerBackoff()); - if (kinesisConsumerProperties.getWorkerId() != null) { - adapter.setWorkerId(kinesisConsumerProperties.getWorkerId()); - } - } - else { - adapter = new KclMessageDrivenChannelAdapter(kinesisClientLibConfiguration, amazonKinesisClient, - this.cloudWatchClient, this.dynamoDBClient); - consumerGroup = kinesisClientLibConfiguration.getApplicationName(); - } + KclMessageDrivenChannelAdapter adapter = + new KclMessageDrivenChannelAdapter(kinesisClientLibConfiguration, amazonKinesisClient, + this.cloudWatchClient, this.dynamoDBClient); + String consumerGroup = kinesisClientLibConfiguration.getApplicationName(); adapter.setCheckpointMode(kinesisConsumerProperties.getCheckpointMode()); adapter.setCheckpointsInterval(kinesisConsumerProperties.getCheckpointInterval()); @@ -396,7 +379,9 @@ public class KinesisMessageChannelBinder extends return adapter; } - private KinesisClientLibConfiguration obtainKinesisClientLibConfiguration(String stream, String group) { + private KinesisClientLibConfiguration obtainKinesisClientLibConfiguration( + KinesisConsumerProperties properties, String stream, String group) { + KinesisClientLibConfiguration candidate = null; for (KinesisClientLibConfiguration conf : this.kinesisClientLibConfigurations) { if (stream.equals(conf.getStreamName())) { @@ -406,6 +391,81 @@ public class KinesisMessageChannelBinder extends } } } + + if (candidate == null) { + boolean anonymous = !StringUtils.hasText(group); + String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group; + + candidate = new KinesisClientLibConfiguration(consumerGroup, + stream, + null, + null, + InitialPositionInStream.LATEST, + this.awsCredentialsProvider, + null, + null, + KinesisClientLibConfiguration.DEFAULT_FAILOVER_TIME_MILLIS, + properties.getWorkerId() != null ? properties.getWorkerId() : UUID.randomUUID().toString(), + KinesisClientLibConfiguration.DEFAULT_MAX_RECORDS, + properties.getIdleBetweenPolls(), + false, + KinesisClientLibConfiguration.DEFAULT_PARENT_SHARD_POLL_INTERVAL_MILLIS, + KinesisClientLibConfiguration.DEFAULT_SHARD_SYNC_INTERVAL_MILLIS, + KinesisClientLibConfiguration.DEFAULT_CLEANUP_LEASES_UPON_SHARDS_COMPLETION, + new ClientConfiguration(), + new ClientConfiguration(), + new ClientConfiguration(), + properties.getConsumerBackoff(), + KinesisClientLibConfiguration.DEFAULT_METRICS_BUFFER_TIME_MILLIS, + KinesisClientLibConfiguration.DEFAULT_METRICS_MAX_QUEUE_SIZE, + KinesisClientLibConfiguration.DEFAULT_VALIDATE_SEQUENCE_NUMBER_BEFORE_CHECKPOINTING, + null, + KinesisClientLibConfiguration.DEFAULT_SHUTDOWN_GRACE_MILLIS, + KinesisClientLibConfiguration.DEFAULT_DDB_BILLING_MODE, + new SimpleRecordsFetcherFactory(), + Duration.ofMinutes(1).toMillis(), + Duration.ofMinutes(5).toMillis(), + Duration.ofMinutes(30).toMillis()); + + String shardIteratorType = properties.getShardIteratorType(); + + KinesisShardOffset kinesisShardOffset = KinesisShardOffset.latest(); + + if (StringUtils.hasText(shardIteratorType)) { + String[] typeValue = shardIteratorType.split(":", 2); + ShardIteratorType iteratorType = ShardIteratorType.valueOf(typeValue[0]); + kinesisShardOffset = new KinesisShardOffset(iteratorType); + if (typeValue.length > 1) { + if (ShardIteratorType.AT_TIMESTAMP.equals(iteratorType)) { + kinesisShardOffset + .setTimestamp(new Date(Long.parseLong(typeValue[1]))); + } + else { + kinesisShardOffset.setSequenceNumber(typeValue[1]); + } + } + } + + kinesisShardOffset = + anonymous || StringUtils.hasText(shardIteratorType) + ? kinesisShardOffset + : KinesisShardOffset.trimHorizon(); + + if (kinesisShardOffset.getIteratorType().equals(ShardIteratorType.AT_TIMESTAMP)) { + candidate.withTimestampAtInitialPositionInStream(kinesisShardOffset.getTimestamp()); + } + else if (kinesisShardOffset.getIteratorType().equals(ShardIteratorType.AT_SEQUENCE_NUMBER) || + kinesisShardOffset.getIteratorType().equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER)) { + + throw new IllegalArgumentException("The KCL does not support 'AT_SEQUENCE_NUMBER' " + + "or 'AFTER_SEQUENCE_NUMBER' initial position in stream."); + } + else { + candidate.withInitialPositionInStream( + InitialPositionInStream.valueOf(kinesisShardOffset.getIteratorType().name())); + } + } + return candidate; } diff --git a/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java b/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java index 22bfba601..b6fbbb6b8 100644 --- a/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java +++ b/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java @@ -27,8 +27,11 @@ import java.util.concurrent.atomic.AtomicReference; import com.amazonaws.SDKGlobalConfiguration; import com.amazonaws.handlers.AsyncHandler; +import com.amazonaws.services.cloudwatch.AmazonCloudWatch; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; import com.amazonaws.services.kinesis.AmazonKinesisAsync; +import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream; +import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration; import com.amazonaws.services.kinesis.model.DescribeStreamRequest; import com.amazonaws.services.kinesis.model.DescribeStreamResult; import com.amazonaws.services.kinesis.model.PutRecordRequest; @@ -57,7 +60,9 @@ import org.springframework.cloud.stream.binder.kinesis.properties.KinesisConsume import org.springframework.cloud.stream.binder.kinesis.properties.KinesisProducerProperties; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.context.ApplicationContext; +import org.springframework.context.Lifecycle; import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChannelAdapter; import org.springframework.integration.aws.inbound.kinesis.KinesisShardOffset; import org.springframework.integration.aws.inbound.kinesis.ListenerMode; import org.springframework.integration.aws.support.AwsRequestFailureException; @@ -95,6 +100,8 @@ public class KinesisBinderTests extends private static AmazonDynamoDBAsync DYNAMO_DB; + private static AmazonCloudWatch CLOUD_WATCH; + public KinesisBinderTests() { this.timeoutMultiplier = 10D; @@ -104,6 +111,7 @@ public class KinesisBinderTests extends public static void setup() { AMAZON_KINESIS = LocalstackContainerTest.kinesisClient(); DYNAMO_DB = LocalstackContainerTest.dynamoDbClient(); + CLOUD_WATCH = LocalstackContainerTest.cloudWatchClient(); System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true"); } @@ -322,6 +330,50 @@ public class KinesisBinderTests extends inputBinding.unbind(); } + @Test + public void testKclWithTimestampAtInitialPositionInStream() throws Exception { + KinesisBinderConfigurationProperties configurationProperties = new KinesisBinderConfigurationProperties(); + configurationProperties.setKplKclEnabled(true); + KinesisTestBinder binder = getBinder(configurationProperties); + DirectChannel output = createBindableChannel("output", new BindingProperties()); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + Date testDate = new Date(); + consumerProperties.getExtension() + .setShardIteratorType(ShardIteratorType.AT_TIMESTAMP.name() + ":" + testDate.getTime()); + Binding binding = binder.bindConsumer("testKclStream", "test", output, consumerProperties); + + Lifecycle lifecycle = TestUtils.getPropertyValue(binding, "lifecycle", Lifecycle.class); + assertThat(lifecycle).isInstanceOf(KclMessageDrivenChannelAdapter.class); + + KinesisClientLibConfiguration config = + TestUtils.getPropertyValue(lifecycle, "config", KinesisClientLibConfiguration.class); + + assertThat(config.getInitialPositionInStream()).isEqualTo(InitialPositionInStream.AT_TIMESTAMP); + assertThat(config.getTimestampAtInitialPositionInStream()).isEqualTo(testDate); + + binding.unbind(); + } + + @Test + public void testKclWithTrimHorizonInitialPositionInStream() throws Exception { + KinesisBinderConfigurationProperties configurationProperties = new KinesisBinderConfigurationProperties(); + configurationProperties.setKplKclEnabled(true); + KinesisTestBinder binder = getBinder(configurationProperties); + DirectChannel output = createBindableChannel("output", new BindingProperties()); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + Binding binding = binder.bindConsumer("testKclStream", null, output, consumerProperties); + + Lifecycle lifecycle = TestUtils.getPropertyValue(binding, "lifecycle", Lifecycle.class); + assertThat(lifecycle).isInstanceOf(KclMessageDrivenChannelAdapter.class); + + KinesisClientLibConfiguration config = + TestUtils.getPropertyValue(lifecycle, "config", KinesisClientLibConfiguration.class); + + assertThat(config.getInitialPositionInStream()).isEqualTo(InitialPositionInStream.TRIM_HORIZON); + + binding.unbind(); + } + @Test @Disabled("Localstack doesn't support updateShardCount. Test only against real AWS Kinesis") public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Exception { @@ -405,7 +457,8 @@ public class KinesisBinderTests extends private KinesisTestBinder getBinder( KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties) { if (this.testBinder == null) { - this.testBinder = new KinesisTestBinder(AMAZON_KINESIS, DYNAMO_DB, kinesisBinderConfigurationProperties); + this.testBinder = new KinesisTestBinder(AMAZON_KINESIS, DYNAMO_DB, CLOUD_WATCH, + kinesisBinderConfigurationProperties); this.timeoutMultiplier = 20; } return this.testBinder; diff --git a/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java b/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java index a6b7a33f1..d2d751c16 100644 --- a/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java +++ b/binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java @@ -16,10 +16,12 @@ package org.springframework.cloud.stream.binder.kinesis; +import java.util.ArrayList; import java.util.List; import com.amazonaws.auth.AWSStaticCredentialsProvider; import com.amazonaws.auth.BasicAWSCredentials; +import com.amazonaws.services.cloudwatch.AmazonCloudWatch; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; import com.amazonaws.services.kinesis.AmazonKinesisAsync; import com.amazonaws.services.kinesis.model.ListStreamsRequest; @@ -40,6 +42,7 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageProducer; @@ -58,7 +61,7 @@ public class KinesisTestBinder extends private final GenericApplicationContext applicationContext; public KinesisTestBinder(AmazonKinesisAsync amazonKinesis, AmazonDynamoDBAsync dynamoDbClient, - KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties) { + AmazonCloudWatch cloudWatchClient, KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties) { this.applicationContext = new AnnotationConfigApplicationContext(Config.class); @@ -68,10 +71,11 @@ public class KinesisTestBinder extends amazonKinesis, kinesisBinderConfigurationProperties); KinesisMessageChannelBinder binder = new TestKinesisMessageChannelBinder( - amazonKinesis, dynamoDbClient, kinesisBinderConfigurationProperties, + amazonKinesis, dynamoDbClient, cloudWatchClient, kinesisBinderConfigurationProperties, provisioningProvider); binder.setApplicationContext(this.applicationContext); + binder.setKinesisClientLibConfigurations(new ArrayList<>()); setBinder(binder); } @@ -136,11 +140,12 @@ public class KinesisTestBinder extends TestKinesisMessageChannelBinder(AmazonKinesisAsync amazonKinesis, AmazonDynamoDBAsync dynamoDbClient, + AmazonCloudWatch cloudWatchClient, KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties, KinesisStreamProvisioner provisioningProvider) { super(kinesisBinderConfigurationProperties, provisioningProvider, amazonKinesis, - new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")), dynamoDbClient, null, null); + new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")), dynamoDbClient, null, cloudWatchClient); } /* @@ -161,10 +166,12 @@ public class KinesisTestBinder extends MessageProducer messageProducer = super.createConsumerEndpoint(destination, group, properties); - DirectFieldAccessor dfa = new DirectFieldAccessor(messageProducer); - dfa.setPropertyValue("describeStreamBackoff", 10); - dfa.setPropertyValue("consumerBackoff", 10); - dfa.setPropertyValue("idleBetweenPolls", 1); + if (messageProducer instanceof KinesisMessageDrivenChannelAdapter) { + DirectFieldAccessor dfa = new DirectFieldAccessor(messageProducer); + dfa.setPropertyValue("describeStreamBackoff", 10); + dfa.setPropertyValue("consumerBackoff", 10); + dfa.setPropertyValue("idleBetweenPolls", 1); + } return messageProducer; }