From dd9f3a7925f96471075f5830768ce2020e2c93d6 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 13 Mar 2019 10:30:10 -0400 Subject: [PATCH] Downgrade to KCL 1.9.3 * Upgrade to DynamoDB Lock Client 1.1 * Use current releases for SI and SC-AWS * Fix compatibility with current SI --- build.gradle | 12 +- .../KclMessageDrivenChannelAdapter.java | 321 +++++++++--------- .../outbound/AbstractAwsMessageHandler.java | 15 +- .../aws/outbound/KinesisMessageHandler.java | 15 +- .../aws/outbound/KplMessageHandler.java | 15 +- .../aws/kinesis/KinesisIntegrationTests.java | 2 +- 6 files changed, 198 insertions(+), 182 deletions(-) diff --git a/build.gradle b/build.gradle index 3f27e73..25a4501 100644 --- a/build.gradle +++ b/build.gradle @@ -32,13 +32,13 @@ repositories { ext { assertjVersion = '3.12.0' awaitilityVersion = '3.1.6' - dynamodbLockClientVersion = '1.0.0' + dynamodbLockClientVersion = '1.1.0' jacksonVersion = '2.9.8' servletApiVersion = '4.0.1' log4jVersion = '2.11.2' - springCloudAwsVersion = '2.1.1.BUILD-SNAPSHOT' - springIntegrationVersion = '5.2.0.BUILD-SNAPSHOT' - kinesisClientVersion = '2.0.5' + springCloudAwsVersion = '2.1.1.RELEASE' + springIntegrationVersion = '5.1.3.RELEASE' + kinesisClientVersion = '1.9.3' kinesisProducerVersion = '0.12.11' idPrefix = 'aws' @@ -95,7 +95,7 @@ dependencies { compile('org.springframework.integration:spring-integration-file', optional) compile('org.springframework.integration:spring-integration-http', optional) - compile("software.amazon.kinesis:amazon-kinesis-client:$kinesisClientVersion", optional) + compile("com.amazonaws:amazon-kinesis-client:$kinesisClientVersion", optional) compile("com.amazonaws:amazon-kinesis-producer:$kinesisProducerVersion", optional) compile('com.amazonaws:aws-java-sdk-kinesis', optional) @@ -110,7 +110,7 @@ dependencies { testCompile ("org.awaitility:awaitility:$awaitilityVersion") { exclude group: 'org.hamcrest' } - + testRuntime "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion" testRuntime "org.apache.logging.log4j:log4j-jcl:$log4jVersion" } 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 23175eb..4f356c7 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 @@ -19,10 +19,11 @@ package org.springframework.integration.aws.inbound.kinesis; import java.nio.ByteBuffer; import java.util.List; import java.util.UUID; -import java.util.concurrent.Executor; import org.springframework.beans.factory.DisposableBean; import org.springframework.core.AttributeAccessor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.core.task.support.ExecutorServiceAdapter; import org.springframework.integration.aws.support.AwsHeaders; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.mapping.InboundMessageMapper; @@ -34,26 +35,28 @@ import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.util.Assert; -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.kinesis.common.ConfigsBuilder; -import software.amazon.kinesis.common.InitialPositionInStream; -import software.amazon.kinesis.common.InitialPositionInStreamExtended; -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.events.InitializationInput; -import software.amazon.kinesis.lifecycle.events.LeaseLostInput; -import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; -import software.amazon.kinesis.lifecycle.events.ShardEndedInput; -import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; -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 com.amazonaws.ClientConfiguration; +import com.amazonaws.regions.Regions; +import com.amazonaws.services.cloudwatch.AmazonCloudWatch; +import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient; +import com.amazonaws.services.cloudwatch.AmazonCloudWatchClientBuilder; +import com.amazonaws.services.dynamodbv2.AmazonDynamoDB; +import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient; +import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder; +import com.amazonaws.services.kinesis.AmazonKinesis; +import com.amazonaws.services.kinesis.AmazonKinesisClient; +import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder; +import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException; +import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException; +import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException; +import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; +import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer; +import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; +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.ShutdownReason; +import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker; +import com.amazonaws.services.kinesis.model.Record; /** * The {@link MessageProducerSupport} implementation for receiving data from Amazon @@ -76,18 +79,17 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple private InboundMessageMapper embeddedHeadersMapper; - private Scheduler scheduler; + private Worker scheduler; - private final Executor executor; + private final TaskExecutor executor; - private final KinesisAsyncClient kinesisClient; + private final AmazonKinesis kinesisClient; - private final CloudWatchAsyncClient cloudWatchClient; + private final AmazonCloudWatch cloudWatchClient; - private final DynamoDbAsyncClient dynamoDBClient; + private final AmazonDynamoDB dynamoDBClient; - private InitialPositionInStreamExtended streamInitialSequence = - InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.LATEST); + private InitialPositionInStream streamInitialSequence = InitialPositionInStream.LATEST; private int idleBetweenPolls; @@ -95,18 +97,21 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple private long checkpointsInterval = 60_000L; - public KclMessageDrivenChannelAdapter(String streams, Executor executor) { - this(streams, executor, KinesisAsyncClient.builder().build(), - CloudWatchAsyncClient.builder().build(), DynamoDbAsyncClient.builder().build()); + public KclMessageDrivenChannelAdapter(String streams, TaskExecutor executor) { + this(streams, executor, AmazonKinesisClientBuilder.defaultClient(), + AmazonCloudWatchClientBuilder.defaultClient(), AmazonDynamoDBClientBuilder.defaultClient()); } - public KclMessageDrivenChannelAdapter(String streams, Executor executor, Region region) { - this(streams, executor, KinesisAsyncClient.builder().region(region).build(), - CloudWatchAsyncClient.builder().region(region).build(), DynamoDbAsyncClient.builder().region(region).build()); + public KclMessageDrivenChannelAdapter(String streams, TaskExecutor executor, Regions region) { + this(streams, executor, AmazonKinesisClient.builder().withRegion(region).build(), + AmazonCloudWatchClient.builder().withRegion(region).build(), + AmazonDynamoDBClient.builder().withRegion(region).build()); } - public KclMessageDrivenChannelAdapter(String stream, Executor executor, - KinesisAsyncClient kinesisClient, CloudWatchAsyncClient cloudWatchClient, DynamoDbAsyncClient dynamoDBClient) { + public KclMessageDrivenChannelAdapter(String stream, TaskExecutor executor, + AmazonKinesis kinesisClient, AmazonCloudWatch cloudWatchClient, + AmazonDynamoDB dynamoDBClient) { + Assert.notNull(stream, "'stream' must not be null."); Assert.notNull(executor, "'executor' must not be null."); Assert.notNull(kinesisClient, "'kinesisClient' must not be null."); @@ -135,26 +140,65 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple this.embeddedHeadersMapper = embeddedHeadersMapper; } + public void setStreamInitialSequence(InitialPositionInStream streamInitialSequence) { + Assert.notNull(streamInitialSequence, "'streamInitialSequence' must not be null"); + this.streamInitialSequence = streamInitialSequence; + } + + public void setIdleBetweenPolls(int idleBetweenPolls) { + this.idleBetweenPolls = Math.max(250, idleBetweenPolls); + } + + public void setConsumerBackoff(int consumerBackoff) { + this.consumerBackoff = Math.max(1000, consumerBackoff); + } + + /** + * Sets the interval between 2 checkpoints. + * + * @param checkpointsInterval interval between 2 checkpoints (in milliseconds) + */ + public void setCheckpointsInterval(long checkpointsInterval) { + this.checkpointsInterval = checkpointsInterval; + } + @Override protected void onInit() { super.onInit(); - String workerId = UUID.randomUUID().toString(); - RecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory(); + KinesisClientLibConfiguration config = + new KinesisClientLibConfiguration( + this.consumerGroup, + this.stream, + null, + this.streamInitialSequence, + null, null, null, + KinesisClientLibConfiguration.DEFAULT_FAILOVER_TIME_MILLIS, + UUID.randomUUID().toString(), + KinesisClientLibConfiguration.DEFAULT_MAX_RECORDS, + this.idleBetweenPolls, + 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(), + this.consumerBackoff, + 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); - ConfigsBuilder configsBuilder = new ConfigsBuilder(this.stream, this.consumerGroup, - this.kinesisClient, this.dynamoDBClient, this.cloudWatchClient, workerId, recordProcessorFactory); - configsBuilder.retrievalConfig().initialPositionInStreamExtended(this.streamInitialSequence); - configsBuilder.retrievalConfig().listShardsBackoffTimeInMillis(this.consumerBackoff); - configsBuilder.coordinatorConfig().parentShardPollIntervalMillis(this.idleBetweenPolls); - - this.scheduler = new Scheduler(configsBuilder.checkpointConfig(), - configsBuilder.coordinatorConfig(), - configsBuilder.leaseManagementConfig(), - configsBuilder.lifecycleConfig(), - configsBuilder.metricsConfig(), - configsBuilder.processorConfig(), - configsBuilder.retrievalConfig()); + this.scheduler = new Worker.Builder() + .kinesisClient(this.kinesisClient) + .dynamoDBClient(this.dynamoDBClient) + .cloudWatchClient(this.cloudWatchClient) + .recordProcessorFactory(new RecordProcessorFactory()) + .execService(new ExecutorServiceAdapter(this.executor)) + .config(config) + .build(); } @Override @@ -186,99 +230,43 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple } } - public void setStreamInitialSequence(InitialPositionInStream streamInitialSequence) { - setStreamInitialSequenceExtended(InitialPositionInStreamExtended.newInitialPosition(streamInitialSequence)); - } - - public void setStreamInitialSequenceExtended(InitialPositionInStreamExtended streamInitialSequence) { - Assert.notNull(streamInitialSequence, "'streamInitialSequence' must not be null"); - this.streamInitialSequence = streamInitialSequence; - } - - public void setIdleBetweenPolls(int idleBetweenPolls) { - this.idleBetweenPolls = Math.max(250, idleBetweenPolls); - } - - public void setConsumerBackoff(int consumerBackoff) { - this.consumerBackoff = Math.max(1000, consumerBackoff); - } - - /** - * Sets the interval between 2 checkpoints. - * - * @param checkpointsInterval interval between 2 checkpoints (in milliseconds) - */ - public void setCheckpointsInterval(long checkpointsInterval) { - this.checkpointsInterval = checkpointsInterval; - } - @Override public String toString() { return "KclMessageDrivenChannelAdapter{consumerGroup='" + this.consumerGroup + '\'' + ", stream='" + this.stream + "'}"; } - private class RecordProcessorFactory implements ShardRecordProcessorFactory { + private class RecordProcessorFactory implements IRecordProcessorFactory { + @Override - public ShardRecordProcessor shardRecordProcessor() { + public IRecordProcessor createProcessor() { return new RecordProcessor(); } + } /** * Processes records and checkpoints progress. */ - private class RecordProcessor implements ShardRecordProcessor { + private class RecordProcessor implements IRecordProcessor { private String shardId; private long nextCheckpointTimeInMillis; - /** {@inheritDoc} */ @Override - public void initialize(InitializationInput initializationInput) { - this.shardId = initializationInput.shardId(); + public void initialize(String shardId) { + this.shardId = shardId; if (logger.isInfoEnabled()) { logger.info("Initializing record processor for shard: " + this.shardId); } } - /** {@inheritDoc} */ @Override - public void leaseLost(LeaseLostInput leaseLostInput) { - logger.info("Lost lease, so terminating."); - } - - /** {@inheritDoc} */ - @Override - public void shardEnded(ShardEndedInput shardEndedInput) { - try { - logger.info("Reached shard end checkpointing."); - shardEndedInput.checkpointer().checkpoint(); + public void processRecords(List records, IRecordProcessorCheckpointer checkpointer) { + if (logger.isDebugEnabled()) { + logger.debug("Processing " + records.size() + " records from " + this.shardId); } - catch (ShutdownException | InvalidStateException e) { - logger.error("Exception while checkpointing at shard end. Giving up", e); - } - } - - /** {@inheritDoc} */ - @Override - public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { - try { - logger.info("Scheduler is shutting down, checkpointing."); - shutdownRequestedInput.checkpointer().checkpoint(); - } - catch (ShutdownException | InvalidStateException e) { - logger.error("Exception while checkpointing at requested shutdown. Giving up", e); - } - } - - /** - * Process records. Skip "poison pill" records. - * - * @param records Data records to be processed. - */ - private void processRecords(List records) { - for (KinesisClientRecord record : records) { + for (Record record : records) { try { processSingleRecord(record); } @@ -287,21 +275,56 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple } finally { attributesHolder.remove(); + // Checkpoint once every checkpoint interval. + if (System.currentTimeMillis() > nextCheckpointTimeInMillis) { + checkpoint(checkpointer); + this.nextCheckpointTimeInMillis = System.currentTimeMillis() + checkpointsInterval; + } } } } /** * Process a single record. - * * @param record The record to be processed. */ - private void processSingleRecord(KinesisClientRecord record) { - + private void processSingleRecord(Record record) { // Convert AWS Record in Spring Message. performSend(prepareMessageForRecord(record), record); } + private AbstractIntegrationMessageBuilder prepareMessageForRecord(Record record) { + ByteBuffer data = record.getData(); + byte[] dataArray = new byte[data.remaining()]; + Object payload = dataArray; + data.get(dataArray); + Message messageToUse = null; + + if (KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) { + try { + messageToUse = + KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper.toMessage((byte[]) payload); + + payload = messageToUse.getPayload(); + } + catch (Exception e) { + logger.warn("Could not parse embedded headers. Remain payload untouched.", e); + } + } + + AbstractIntegrationMessageBuilder messageBuilder = getMessageBuilderFactory().withPayload(payload) + .setHeader(AwsHeaders.RECEIVED_PARTITION_KEY, record.getPartitionKey()) + .setHeader(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, record.getSequenceNumber()) + .setHeader(AwsHeaders.RECEIVED_STREAM, KclMessageDrivenChannelAdapter.this.stream) + .setHeader(AwsHeaders.SHARD, this.shardId); + + if (messageToUse != null) { + messageBuilder.copyHeadersIfAbsent(messageToUse.getHeaders()); + } + + return messageBuilder; + } + private void performSend(AbstractIntegrationMessageBuilder messageBuilder, Object rawRecord) { Message messageToSend = messageBuilder.build(); setAttributesIfNecessary(rawRecord, messageToSend); @@ -328,43 +351,11 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple } } - private AbstractIntegrationMessageBuilder prepareMessageForRecord(KinesisClientRecord record) { - ByteBuffer data = record.data(); - byte[] dataArray = new byte[data.remaining()]; - Object payload = dataArray; - data.get(dataArray); - Message messageToUse = null; - - if (KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) { - try { - messageToUse = KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper.toMessage((byte[]) payload); - - payload = messageToUse.getPayload(); - } - catch (Exception e) { - logger.warn("Could not parse embedded headers. Remain payload untouched.", e); - } - } - - AbstractIntegrationMessageBuilder messageBuilder = getMessageBuilderFactory().withPayload(payload) - .setHeader(AwsHeaders.RECEIVED_PARTITION_KEY, record.partitionKey()) - .setHeader(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, record.sequenceNumber()) - .setHeader(AwsHeaders.RECEIVED_STREAM, KclMessageDrivenChannelAdapter.this.stream) - .setHeader(AwsHeaders.SHARD, this.shardId); - - if (messageToUse != null) { - messageBuilder.copyHeadersIfAbsent(messageToUse.getHeaders()); - } - - return messageBuilder; - } - /** * Checkpoint with retries. - * * @param checkpointer checkpointer */ - private void checkpoint(RecordProcessorCheckpointer checkpointer) { + private void checkpoint(IRecordProcessorCheckpointer checkpointer) { if (logger.isInfoEnabled()) { logger.info("Checkpointing shard " + shardId); } @@ -389,20 +380,18 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple } @Override - public void processRecords(ProcessRecordsInput processRecordsInput) { - List records = processRecordsInput.records(); - if (logger.isDebugEnabled()) { - logger.debug("Processing " + records.size() + " records from " + this.shardId); + public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) { + if (logger.isInfoEnabled()) { + logger.info("Scheduler is shutting down for reason '" + reason + "'; checkpointing..."); } - - // Process records and perform all exception handling. - processRecords(records); - - // Checkpoint once every checkpoint interval. - if (System.currentTimeMillis() > nextCheckpointTimeInMillis) { - checkpoint(processRecordsInput.checkpointer()); - this.nextCheckpointTimeInMillis = System.currentTimeMillis() + checkpointsInterval; + try { + checkpointer.checkpoint(); + } + catch (ShutdownException | InvalidStateException e) { + logger.error("Exception while checkpointing at requested shutdown. Giving up", e); } } + } + } diff --git a/src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java index 0eaccd7..6a50f6f 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java @@ -16,6 +16,7 @@ package org.springframework.integration.aws.outbound; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -173,13 +174,18 @@ public abstract class AbstractAwsMessageHandler extends AbstractMessageProduc } @Override - protected void handleMessageInternal(Message message) throws Exception { + protected void handleMessageInternal(Message message) { Future resultFuture = handleMessageToAws(message); if (this.sync) { Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class); if (sendTimeout == null || sendTimeout < 0) { - resultFuture.get(); + try { + resultFuture.get(); + } + catch (InterruptedException | ExecutionException ex) { + throw new IllegalStateException(ex); + } } else { try { @@ -188,6 +194,9 @@ public abstract class AbstractAwsMessageHandler extends AbstractMessageProduc catch (TimeoutException te) { throw new MessageTimeoutException(message, "Timeout waiting for response from AmazonKinesis", te); } + catch (InterruptedException | ExecutionException ex) { + throw new IllegalStateException(ex); + } } } } @@ -234,7 +243,7 @@ public abstract class AbstractAwsMessageHandler extends AbstractMessageProduc }; } - protected abstract Future handleMessageToAws(Message message) throws Exception; + protected abstract Future handleMessageToAws(Message message); protected abstract void additionalOnSuccessHeaders(AbstractIntegrationMessageBuilder messageBuilder, AmazonWebServiceRequest request, Object result); diff --git a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java index 0b297b5..94086a6 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java @@ -31,6 +31,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder import org.springframework.integration.support.MutableMessage; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -152,7 +153,7 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler { } @Override - protected Future handleMessageToAws(Message message) throws Exception { + protected Future handleMessageToAws(Message message) { if (message.getPayload() instanceof PutRecordsRequest) { AsyncHandler asyncHandler = obtainAsyncHandler(message, (PutRecordsRequest) message.getPayload()); @@ -172,7 +173,7 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler { } } - private PutRecordRequest buildPutRecordRequest(Message message) throws Exception { + private PutRecordRequest buildPutRecordRequest(Message message) { MessageHeaders messageHeaders = message.getHeaders(); String stream = messageHeaders.get(AwsHeaders.STREAM, String.class); if (!StringUtils.hasText(stream) && this.streamExpression != null) { @@ -217,6 +218,7 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler { payload instanceof byte[] ? (byte[]) payload : this.converter.convert(payload); + Assert.notNull(bytes, "payload cannot be null"); if (this.embeddedHeadersMapper != null) { messageToEmbed = new MutableMessage<>(bytes, messageHeaders); } @@ -226,7 +228,14 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler { } if (messageToEmbed != null) { - data = ByteBuffer.wrap(this.embeddedHeadersMapper.fromMessage(messageToEmbed)); + try { + byte[] bytes = this.embeddedHeadersMapper.fromMessage(messageToEmbed); + Assert.notNull(bytes, "payload cannot be null"); + data = ByteBuffer.wrap(bytes); + } + catch (Exception ex) { + throw new MessageConversionException(message, "Cannot embedded headers to payload", ex); + } } return new PutRecordRequest() diff --git a/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java index 5d7e576..d261e9e 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java @@ -32,6 +32,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder import org.springframework.integration.support.MutableMessage; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -159,7 +160,7 @@ public class KplMessageHandler extends AbstractAwsMessageHandler { } @Override - protected Future handleMessageToAws(Message message) throws Exception { + protected Future handleMessageToAws(Message message) { if (message.getPayload() instanceof PutRecordsRequest) { throw new UnsupportedOperationException("not implemented"); } @@ -208,7 +209,7 @@ public class KplMessageHandler extends AbstractAwsMessageHandler { return recordResult; } - private PutRecordRequest buildPutRecordRequest(Message message) throws Exception { + private PutRecordRequest buildPutRecordRequest(Message message) { MessageHeaders messageHeaders = message.getHeaders(); String stream = messageHeaders.get(AwsHeaders.STREAM, String.class); if (!StringUtils.hasText(stream) && this.streamExpression != null) { @@ -253,6 +254,7 @@ public class KplMessageHandler extends AbstractAwsMessageHandler { payload instanceof byte[] ? (byte[]) payload : this.converter.convert(payload); + Assert.notNull(bytes, "payload cannot be null"); if (this.embeddedHeadersMapper != null) { messageToEmbed = new MutableMessage<>(bytes, messageHeaders); } @@ -262,7 +264,14 @@ public class KplMessageHandler extends AbstractAwsMessageHandler { } if (messageToEmbed != null) { - data = ByteBuffer.wrap(this.embeddedHeadersMapper.fromMessage(messageToEmbed)); + try { + byte[] bytes = this.embeddedHeadersMapper.fromMessage(messageToEmbed); + Assert.notNull(bytes, "payload cannot be null"); + data = ByteBuffer.wrap(bytes); + } + catch (Exception ex) { + throw new MessageConversionException(message, "Cannot embedded headers to payload", ex); + } } return new PutRecordRequest() diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java index 38595d8..fe26534 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -112,7 +112,7 @@ public class KinesisIntegrationTests { assertThat(errorMessage).isNotNull(); assertThat(errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD)).isNotNull(); assertThat(((Exception) errorMessage.getPayload()).getMessage()) - .contains("Channel 'kinesisReceiveChannel' expected one of the following datataypes " + + .contains("Channel 'kinesisReceiveChannel' expected one of the following data types " + "[class java.util.Date], but received [class java.lang.String]");