SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2359: Enable other KCL initial offsets

Browse Source 
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
This commit is contained in:
Artem Bilan
2022-04-27 12:54:05 -04:00
committed by Oleg Zhurakousky
parent 210eb7e9f5
commit d8fd4c2aab
3 changed files with 156 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java
View File

@@ -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;
}

55
binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java
View File

@@ -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<KinesisConsumerProperties> 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<KinesisConsumerProperties> 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;

21
binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java
View File

@@ -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;
}