Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix KCL channel adapter

Browse Source 
* 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
This commit is contained in:
abilan
2023-03-16 15:45:23 -04:00
parent 1861a98625
commit f7bec6a4f5
3 changed files with 142 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java
View File

@@ -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<Object> 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<Object> messageBuilder = getMessageBuilderFactory().withPayload(payload)
.setHeader(AwsHeaders.RECEIVED_PARTITION_KEY, record.partitionKey())
.setHeader(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, record.sequenceNumber());
AbstractIntegrationMessageBuilder<Object> 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);

122
src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java Normal file
View File

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

6
src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java
View File

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