From f7bec6a4f5cefc4e5383ec7311b98e73d1a84637 Mon Sep 17 00:00:00 2001 From: abilan Date: Thu, 16 Mar 2023 15:45:23 -0400 Subject: [PATCH] Fix KCL channel adapter * Fix config mutation and its propagation down to the `Scheduler` * Fix conversion from `ByteBuffer` * Add `KclMessageDrivenChannelAdapterTests` to verify KCL in real action. The test is slow enough because KCL has a long initialization phase * Disable `KplKclIntegrationTests` back because KPL native daemon makes some real calls to EC2 and does not understand credentials from Testcontainers --- .../KclMessageDrivenChannelAdapter.java | 29 +++-- .../KclMessageDrivenChannelAdapterTests.java | 122 ++++++++++++++++++ .../aws/kinesis/KplKclIntegrationTests.java | 6 +- 3 files changed, 142 insertions(+), 15 deletions(-) create mode 100644 src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java index 085b72f..3a97c5c 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java @@ -30,6 +30,7 @@ import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.utils.BinaryUtils; import software.amazon.kinesis.common.ConfigsBuilder; import software.amazon.kinesis.common.InitialPositionInStream; import software.amazon.kinesis.common.InitialPositionInStreamExtended; @@ -39,6 +40,7 @@ import software.amazon.kinesis.coordinator.Scheduler; import software.amazon.kinesis.exceptions.InvalidStateException; import software.amazon.kinesis.exceptions.ShutdownException; import software.amazon.kinesis.exceptions.ThrottlingException; +import software.amazon.kinesis.lifecycle.LifecycleConfig; import software.amazon.kinesis.lifecycle.events.InitializationInput; import software.amazon.kinesis.lifecycle.events.LeaseLostInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; @@ -50,6 +52,7 @@ import software.amazon.kinesis.processor.RecordProcessorCheckpointer; import software.amazon.kinesis.processor.ShardRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessorFactory; import software.amazon.kinesis.retrieval.KinesisClientRecord; +import software.amazon.kinesis.retrieval.RetrievalConfig; import org.springframework.core.AttributeAccessor; import org.springframework.core.convert.converter.Converter; @@ -243,9 +246,6 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { this.config = new ConfigsBuilder(new StreamsTracker(), this.consumerGroup, this.kinesisClient, this.dynamoDBClient, this.cloudWatchClient, this.workerId, this.recordProcessorFactory); } - - this.config.lifecycleConfig().taskBackoffTimeMillis(this.consumerBackoff); - this.config.retrievalConfig().glueSchemaRegistryDeserializer(this.glueSchemaRegistryDeserializer); } @Override @@ -258,15 +258,21 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { + "because it does not make sense in case of [ListenerMode.batch]."); } + LifecycleConfig lifecycleConfig = this.config.lifecycleConfig().taskBackoffTimeMillis(this.consumerBackoff); + RetrievalConfig retrievalConfig = + this.config.retrievalConfig() + .glueSchemaRegistryDeserializer(this.glueSchemaRegistryDeserializer) + .initialPositionInStreamExtended(this.streamInitialSequence); + this.scheduler = new Scheduler( this.config.checkpointConfig(), this.config.coordinatorConfig(), this.config.leaseManagementConfig(), - this.config.lifecycleConfig(), + lifecycleConfig, this.config.metricsConfig(), this.config.processorConfig(), - this.config.retrievalConfig()); + retrievalConfig); this.executor.execute(this.scheduler); } @@ -467,13 +473,12 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { } private AbstractIntegrationMessageBuilder prepareMessageForRecord(KinesisClientRecord record) { - Object payload = record.data().array(); + Object payload = BinaryUtils.copyAllBytesFrom(record.data()); Message messageToUse = null; if (KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) { try { - messageToUse = KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper - .toMessage((byte[]) payload); + messageToUse = KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper.toMessage((byte[]) payload); if (messageToUse == null) { throw new IllegalStateException("The 'embeddedHeadersMapper' returned null for payload: " + Arrays.toString((byte[]) payload)); @@ -489,9 +494,11 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { payload = KclMessageDrivenChannelAdapter.this.converter.convert((byte[]) payload); } - AbstractIntegrationMessageBuilder messageBuilder = getMessageBuilderFactory().withPayload(payload) - .setHeader(AwsHeaders.RECEIVED_PARTITION_KEY, record.partitionKey()) - .setHeader(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, record.sequenceNumber()); + AbstractIntegrationMessageBuilder messageBuilder = + getMessageBuilderFactory() + .withPayload(payload) + .setHeader(AwsHeaders.RECEIVED_PARTITION_KEY, record.partitionKey()) + .setHeader(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, record.sequenceNumber()); if (KclMessageDrivenChannelAdapter.this.bindSourceRecord) { messageBuilder.setHeader(IntegrationMessageHeaderAccessor.SOURCE_DATA, record); diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java new file mode 100644 index 0000000..ff24b6c --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java @@ -0,0 +1,122 @@ +/* + * Copyright 2023 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License 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 org.springframework.integration.aws.kinesis; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.SdkBytes; +import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.kinesis.common.InitialPositionInStream; +import software.amazon.kinesis.common.InitialPositionInStreamExtended; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.aws.LocalstackContainerTest; +import org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChannelAdapter; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Artem Bilan + * + * @since 3.0 + */ +@SpringJUnitConfig +@DirtiesContext +public class KclMessageDrivenChannelAdapterTests implements LocalstackContainerTest { + + private static final String TEST_STREAM = "TestStreamKcl"; + + private static KinesisAsyncClient AMAZON_KINESIS; + + private static DynamoDbAsyncClient DYNAMO_DB; + + private static CloudWatchAsyncClient CLOUD_WATCH; + + @Autowired + private PollableChannel kinesisReceiveChannel; + + @BeforeAll + static void setup() { + AMAZON_KINESIS = LocalstackContainerTest.kinesisClient(); + DYNAMO_DB = LocalstackContainerTest.dynamoDbClient(); + CLOUD_WATCH = LocalstackContainerTest.cloudWatchClient(); + + AMAZON_KINESIS.createStream(request -> request.streamName(TEST_STREAM).shardCount(1)) + .thenCompose(result -> + AMAZON_KINESIS.waiter().waitUntilStreamExists(request -> request.streamName(TEST_STREAM))) + .join(); + } + + @AfterAll + static void tearDown() { + AMAZON_KINESIS.deleteStream(request -> request.streamName(TEST_STREAM)); + } + + @Test + void kclChannelAdapterReceivesRecords() { + String testData = "test data"; + + AMAZON_KINESIS.putRecord(request -> + request.streamName(TEST_STREAM) + .data(SdkBytes.fromUtf8String(testData)) + .partitionKey("test")); + + // We need so long delay because KCL has a more than a minute setup phase. + Message receive = this.kinesisReceiveChannel.receive(120_000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo(testData); + assertThat(receive.getHeaders()).containsKey(IntegrationMessageHeaderAccessor.SOURCE_DATA); + assertThat(receive.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, String.class)).isNotEmpty(); + } + + @Configuration + @EnableIntegration + public static class TestConfiguration { + + @Bean + public KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter() { + KclMessageDrivenChannelAdapter adapter = + new KclMessageDrivenChannelAdapter(AMAZON_KINESIS, CLOUD_WATCH, DYNAMO_DB, TEST_STREAM); + adapter.setOutputChannel(kinesisReceiveChannel()); + adapter.setStreamInitialSequence( + InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON)); + adapter.setConverter(String::new); + adapter.setBindSourceRecord(true); + return adapter; + } + + @Bean + public PollableChannel kinesisReceiveChannel() { + return new QueueChannel(); + } + + } + +} diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java index 5a76114..ac45e9a 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java @@ -25,9 +25,8 @@ import com.amazonaws.services.kinesis.producer.KinesisProducer; import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.testcontainers.containers.localstack.LocalStackContainer; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; @@ -66,7 +65,7 @@ import static org.assertj.core.api.Assertions.entry; * * @since 1.1 */ -@DisabledOnOs(OS.WINDOWS) +@Disabled("Depends on real call to http://169.254.169.254 through native library") @SpringJUnitConfig @DirtiesContext public class KplKclIntegrationTests implements LocalstackContainerTest { @@ -105,7 +104,6 @@ public class KplKclIntegrationTests implements LocalstackContainerTest { AMAZON_KINESIS.deleteStream(request -> request.streamName(TEST_STREAM)); } - @Test void testKinesisInboundOutbound() { this.kinesisSendChannel