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

Add CredentialsProvider for KCP Channel Adapter

Browse Source 
* Since internal `KinesisProxy` instance is fully based on the
`CredentialsProvider` provided in the `KinesisClientLibConfiguration`,
we don't have choice unless expose such an option for the
`KclMessageDrivenChannelAdapter`
* Make it as `DefaultAWSCredentialsProviderChain` for those ctors which
are based on the default client and add one more ctor argument for
externally provided clients
* Move an `executor` option to the setter and rely on the
`SimpleAsyncTaskExecutor` by default
* Do not copy `record.getData()` to a new array since
`record.getData().array()` is fully enough for the current version
This commit is contained in:
Artem Bilan
2019-03-14 10:31:40 -04:00
parent dd9f3a7925
commit b1cad7e55a
2 changed files with 41 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,12 +16,12 @@
package org.springframework.integration.aws.inbound.kinesis;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.AttributeAccessor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.core.task.support.ExecutorServiceAdapter;
import org.springframework.integration.aws.support.AwsHeaders;
@@ -36,6 +36,8 @@ import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.cloudwatch.AmazonCloudWatch;
import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
@@ -69,26 +71,28 @@ import com.amazonaws.services.kinesis.model.Record;
*/
@ManagedResource
@IntegrationManagedResource
public class KclMessageDrivenChannelAdapter extends MessageProducerSupport implements DisposableBean {
public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<>();
private final String stream;
private final AmazonKinesis kinesisClient;
private final AWSCredentialsProvider kinesisProxyCredentialsProvider;
private final AmazonCloudWatch cloudWatchClient;
private final AmazonDynamoDB dynamoDBClient;
private TaskExecutor executor = new SimpleAsyncTaskExecutor();
private String consumerGroup = "SpringIntegration";
private InboundMessageMapper<byte[]> embeddedHeadersMapper;
private Worker scheduler;
private final TaskExecutor executor;
private final AmazonKinesis kinesisClient;
private final AmazonCloudWatch cloudWatchClient;
private final AmazonDynamoDB dynamoDBClient;
private InitialPositionInStream streamInitialSequence = InitialPositionInStream.LATEST;
private int idleBetweenPolls;
@@ -97,31 +101,37 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple
private long checkpointsInterval = 60_000L;
public KclMessageDrivenChannelAdapter(String streams, TaskExecutor executor) {
this(streams, executor, AmazonKinesisClientBuilder.defaultClient(),
AmazonCloudWatchClientBuilder.defaultClient(), AmazonDynamoDBClientBuilder.defaultClient());
public KclMessageDrivenChannelAdapter(String streams) {
this(streams, AmazonKinesisClientBuilder.defaultClient(),
AmazonCloudWatchClientBuilder.defaultClient(), AmazonDynamoDBClientBuilder.defaultClient(),
new DefaultAWSCredentialsProviderChain());
}
public KclMessageDrivenChannelAdapter(String streams, TaskExecutor executor, Regions region) {
this(streams, executor, AmazonKinesisClient.builder().withRegion(region).build(),
public KclMessageDrivenChannelAdapter(String streams, Regions region) {
this(streams, AmazonKinesisClient.builder().withRegion(region).build(),
AmazonCloudWatchClient.builder().withRegion(region).build(),
AmazonDynamoDBClient.builder().withRegion(region).build());
AmazonDynamoDBClient.builder().withRegion(region).build(), new DefaultAWSCredentialsProviderChain());
}
public KclMessageDrivenChannelAdapter(String stream, TaskExecutor executor,
public KclMessageDrivenChannelAdapter(String stream,
AmazonKinesis kinesisClient, AmazonCloudWatch cloudWatchClient,
AmazonDynamoDB dynamoDBClient) {
AmazonDynamoDB dynamoDBClient, AWSCredentialsProvider kinesisProxyCredentialsProvider) {
Assert.notNull(stream, "'stream' must not be null.");
Assert.notNull(executor, "'executor' must not be null.");
Assert.notNull(kinesisClient, "'kinesisClient' must not be null.");
Assert.notNull(cloudWatchClient, "'cloudWatchClient' must not be null.");
Assert.notNull(dynamoDBClient, "'dynamoDBClient' must not be null.");
Assert.notNull(kinesisProxyCredentialsProvider, "'kinesisProxyCredentialsProvider' must not be null.");
this.stream = stream;
this.executor = executor;
this.kinesisClient = kinesisClient;
this.cloudWatchClient = cloudWatchClient;
this.dynamoDBClient = dynamoDBClient;
this.kinesisProxyCredentialsProvider = kinesisProxyCredentialsProvider;
}
public void setExecutor(TaskExecutor executor) {
Assert.notNull(executor, "'executor' must not be null.");
this.executor = executor;
}
public void setConsumerGroup(String consumerGroup) {
@@ -132,9 +142,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple
/**
* Specify an {@link InboundMessageMapper} to extract message headers embedded
* into the record data.
*
* @param embeddedHeadersMapper the {@link InboundMessageMapper} to use.
* @since 2.0
*/
public void setEmbeddedHeadersMapper(InboundMessageMapper<byte[]> embeddedHeadersMapper) {
this.embeddedHeadersMapper = embeddedHeadersMapper;
@@ -172,7 +180,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple
this.stream,
null,
this.streamInitialSequence,
null, null, null,
this.kinesisProxyCredentialsProvider,
null, null,
KinesisClientLibConfiguration.DEFAULT_FAILOVER_TIME_MILLIS,
UUID.randomUUID().toString(),
KinesisClientLibConfiguration.DEFAULT_MAX_RECORDS,
@@ -213,7 +222,6 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple
*/
@Override
protected void doStop() {
super.doStop();
this.scheduler.shutdown();
@@ -232,7 +240,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple
@Override
public String toString() {
return "KclMessageDrivenChannelAdapter{consumerGroup='" + this.consumerGroup + '\'' + ", stream='" + this.stream + "'}";
return "KclMessageDrivenChannelAdapter{consumerGroup='" + this.consumerGroup + '\'' +
", stream='" + this.stream + "'}";
}
private class RecordProcessorFactory implements IRecordProcessorFactory {
@@ -294,17 +303,17 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport imple
}
private AbstractIntegrationMessageBuilder<Object> prepareMessageForRecord(Record record) {
ByteBuffer data = record.getData();
byte[] dataArray = new byte[data.remaining()];
Object payload = dataArray;
data.get(dataArray);
Object payload = record.getData().array();
Message<?> messageToUse = null;
if (KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) {
try {
messageToUse =
KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper.toMessage((byte[]) payload);
if (messageToUse == null) {
throw new IllegalStateException("The 'embeddedHeadersMapper' returned null for payload: " +
Arrays.toString((byte[]) payload));
}
payload = messageToUse.getPayload();
}
catch (Exception e) {

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 data types " +
.contains("Channel 'kinesisReceiveChannel' expected one of the following datataypes " +
"[class java.util.Date], but received [class java.lang.String]");