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

Downgrade to KCL 1.9.3

Browse Source 
* Upgrade to DynamoDB Lock Client 1.1
* Use current releases for SI and SC-AWS
* Fix compatibility with current SI
This commit is contained in:
Artem Bilan
2019-03-13 10:30:10 -04:00
parent 1b671378d7
commit dd9f3a7925
6 changed files with 198 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
build.gradle
View File

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

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

@@ -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<byte[]> 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<Record> 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<KinesisClientRecord> 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<Object> 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<Object> 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<Object> 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<Object> 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<KinesisClientRecord> 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);
}
}
}
}

15
src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java
View File

@@ -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<H> 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<H> 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<H> 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);

15
src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java
View File

@@ -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<Void> {
}
@Override
protected Future<?> handleMessageToAws(Message<?> message) throws Exception {
protected Future<?> handleMessageToAws(Message<?> message) {
if (message.getPayload() instanceof PutRecordsRequest) {
AsyncHandler<PutRecordsRequest, PutRecordsResult> asyncHandler =
obtainAsyncHandler(message, (PutRecordsRequest) message.getPayload());
@@ -172,7 +173,7 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler<Void> {
}
}
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<Void> {
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<Void> {
}
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()

15
src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java
View File

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

2
src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java
View File

@@ -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]");