diff --git a/build.gradle b/build.gradle index 9aabaa2..4d96b4d 100644 --- a/build.gradle +++ b/build.gradle @@ -31,11 +31,11 @@ repositories { } ext { - assertjVersion = '3.8.0' + assertjVersion = '3.9.1' servletApiVersion = '3.1.0' - slf4jVersion = '1.7.25' - springCloudAwsVersion = '2.0.0.M3' - springIntegrationVersion = '5.0.1.RELEASE' + log4jVersion = '2.10.0' + springCloudAwsVersion = '2.0.0.M4' + springIntegrationVersion = '5.0.3.RELEASE' idPrefix = 'aws' @@ -102,6 +102,8 @@ dependencies { compile('org.springframework.integration:spring-integration-file', optional) compile('org.springframework.integration:spring-integration-http', optional) +// compile('com.amazonaws:amazon-kinesis-client:1.9.0', optional) + compile('com.amazonaws:aws-java-sdk-kinesis', optional) compile('com.amazonaws:aws-java-sdk-dynamodb', optional) @@ -110,7 +112,8 @@ dependencies { testCompile 'org.springframework.integration:spring-integration-test' testCompile "org.assertj:assertj-core:$assertjVersion" - testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion" + testRuntime "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion" + testRuntime "org.apache.logging.log4j:log4j-jcl:$log4jVersion" } eclipse.project.natures += 'org.springframework.ide.eclipse.core.springnature' diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/Checkpointer.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/Checkpointer.java index 2d083b3..cba5fcc 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/Checkpointer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/Checkpointer.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -20,12 +20,22 @@ package org.springframework.integration.aws.inbound.kinesis; * A callback for target record process to perform checkpoint on the related shard. * * @author Artem Bilan + * * @since 1.1 */ public interface Checkpointer { - void checkpoint(); + /** + * Checkpoint the currently held sequence number if it is bigger than already stored. + * @return true if checkpoint performed; false otherwise. + */ + boolean checkpoint(); - void checkpoint(String sequenceNumber); + /** + * Checkpoint the provided sequence number, if it is bigger than already stored. + * @param sequenceNumber the sequence number to checkpoint. + * @return true if checkpoint performed; false otherwise. + */ + boolean checkpoint(String sequenceNumber); } diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index cb35ca2..2d0e35f 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -65,7 +65,6 @@ import com.amazonaws.services.kinesis.model.ExpiredIteratorException; import com.amazonaws.services.kinesis.model.GetRecordsRequest; import com.amazonaws.services.kinesis.model.GetRecordsResult; import com.amazonaws.services.kinesis.model.GetShardIteratorRequest; -import com.amazonaws.services.kinesis.model.LimitExceededException; import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException; import com.amazonaws.services.kinesis.model.Record; import com.amazonaws.services.kinesis.model.ResourceNotFoundException; @@ -467,14 +466,14 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i try { describeStreamResult = this.amazonKinesis.describeStream(describeStreamRequest); } - catch (LimitExceededException e) { - logger.info("Got LimitExceededException when describing stream [" + stream + "]. " + - "Backing off for [" + this.describeStreamBackoff + "] millis."); + catch (Exception e) { + logger.info("Got an exception when describing stream [" + stream + "]. " + + "Backing off for [" + this.describeStreamBackoff + "] millis.", e); } if (describeStreamResult == null || - !StreamStatus.ACTIVE.toString() - .equals(describeStreamResult.getStreamDescription().getStreamStatus())) { + !StreamStatus.ACTIVE.toString().equals(describeStreamResult.getStreamDescription().getStreamStatus())) { + if (describeStreamRetries++ > this.describeStreamRetries) { ResourceNotFoundException resourceNotFoundException = new ResourceNotFoundException("The stream [" + stream + @@ -494,34 +493,43 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } List shards = describeStreamResult.getStreamDescription().getShards(); - for (Shard shard : shards) { - String endingSequenceNumber = shard.getSequenceNumberRange().getEndingSequenceNumber(); - if (endingSequenceNumber != null) { - String key = buildCheckpointKeyForShard(stream, shard.getShardId()); - String checkpoint = this.checkpointStore.get(key); - boolean skipClosedShard = checkpoint != null && - new BigInteger(endingSequenceNumber) - .compareTo(new BigInteger(checkpoint)) <= 0; + try { + for (Shard shard : shards) { + String endingSequenceNumber = shard.getSequenceNumberRange().getEndingSequenceNumber(); + if (endingSequenceNumber != null) { + String key = buildCheckpointKeyForShard(stream, shard.getShardId()); + String checkpoint = this.checkpointStore.get(key); - if (logger.isTraceEnabled()) { - logger.trace("The shard [" + shard + "] in stream [" + stream + - "] is closed CLOSED with endingSequenceNumber [" + endingSequenceNumber + - "].\nThe last processed checkpoint is [" + checkpoint + "]." + - (skipClosedShard ? "\nThe shard will be skipped." : "")); + boolean skipClosedShard = checkpoint != null && + new BigInteger(endingSequenceNumber) + .compareTo(new BigInteger(checkpoint)) <= 0; + + if (logger.isTraceEnabled()) { + logger.trace("The shard [" + shard + "] in stream [" + stream + + "] is closed CLOSED with endingSequenceNumber [" + endingSequenceNumber + + "].\nThe last processed checkpoint is [" + checkpoint + "]." + + (skipClosedShard ? "\nThe shard will be skipped." : "")); + } + + if (skipClosedShard) { + // Skip CLOSED shard which has been read before according a checkpoint + continue; + } } - if (skipClosedShard) { - // Skip CLOSED shard which has been read before according a checkpoint - continue; - } + shardsToConsume.add(shard); } - - shardsToConsume.add(shard); + } + catch (Exception e) { + logger.info("Got an exception when processing shards in stream [" + stream + "].\n" + + "Retrying...", e); + continue; } if (describeStreamResult.getStreamDescription().getHasMoreShards()) { exclusiveStartShardId = shards.get(shards.size() - 1).getShardId(); + describeStreamRetries = 0; } else { break; @@ -666,16 +674,16 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } for (Iterator iterator = - KinesisMessageDrivenChannelAdapter.this.shardConsumers - .values() - .iterator(); + KinesisMessageDrivenChannelAdapter.this.shardConsumers.values().iterator(); iterator.hasNext(); ) { + ShardConsumer shardConsumer = iterator.next(); shardConsumer.execute(); if (ConsumerState.STOP == shardConsumer.state) { iterator.remove(); if (KinesisMessageDrivenChannelAdapter.this.streams != null && shardConsumer.shardIterator == null) { + // Shard is CLOSED and we are capable for resharding KinesisShardOffset shardOffset = shardConsumer.shardOffset; String stream = shardOffset.getStream(); @@ -753,30 +761,33 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i case NEW: case EXPIRED: this.task = () -> { - if (logger.isInfoEnabled() && this.state == ConsumerState.NEW) { - logger.info("The [" + this + "] has been started."); - } - if (this.shardOffset.isReset()) { - this.checkpointer.remove(); - } - else { - String checkpoint = this.checkpointer.getCheckpoint(); - if (checkpoint != null) { - this.shardOffset.setSequenceNumber(checkpoint); - this.shardOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER); + try { + if (logger.isInfoEnabled() && this.state == ConsumerState.NEW) { + logger.info("The [" + this + "] has been started."); + } + if (this.shardOffset.isReset()) { + this.checkpointer.remove(); + } + else { + String checkpoint = this.checkpointer.getCheckpoint(); + if (checkpoint != null) { + this.shardOffset.setSequenceNumber(checkpoint); + this.shardOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER); + } + } + + GetShardIteratorRequest shardIteratorRequest = this.shardOffset.toShardIteratorRequest(); + this.shardIterator = + KinesisMessageDrivenChannelAdapter.this.amazonKinesis + .getShardIterator(shardIteratorRequest) + .getShardIterator(); + if (ConsumerState.STOP != this.state) { + this.state = ConsumerState.CONSUME; } } - - GetShardIteratorRequest shardIteratorRequest = this.shardOffset.toShardIteratorRequest(); - this.shardIterator = - KinesisMessageDrivenChannelAdapter.this.amazonKinesis - .getShardIterator(shardIteratorRequest) - .getShardIterator(); - if (ConsumerState.STOP != this.state) { - this.state = ConsumerState.CONSUME; + finally { + this.task = null; } - - this.task = null; }; break; @@ -825,9 +836,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i getRecordsRequest.setShardIterator(this.shardIterator); getRecordsRequest.setLimit(KinesisMessageDrivenChannelAdapter.this.recordsLimit); - GetRecordsResult result = getRecords(getRecordsRequest); + GetRecordsResult result = null; try { + result = getRecords(getRecordsRequest); if (result != null) { List records = result.getRecords(); @@ -895,8 +907,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i ShardConsumer.this.state = ConsumerState.SLEEP; } - private void processRecords(List records) { - records = this.checkpointer.filterRecords(records); + private void processRecords(List recordsToProcess) { + List records = this.checkpointer.filterRecords(recordsToProcess); if (!records.isEmpty()) { if (logger.isTraceEnabled()) { logger.trace("Processing records: " + records + " for [" + ShardConsumer.this + "]"); @@ -904,7 +916,14 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i // TODO Reconsider this logic after rebalance and shard leader election implementation if (CheckpointMode.batch.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) { - this.checkpointer.checkpoint(); + if (!this.checkpointer.checkpoint()) { + if (logger.isInfoEnabled()) { + logger.info("The records '" + recordsToProcess + "' are skipped from processing because " + + "their sequence numbers are less than already checkpointed: " + + this.checkpointer.getCheckpoint()); + } + return; + } } switch (KinesisMessageDrivenChannelAdapter.this.listenerMode) { @@ -925,7 +944,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i messageBuilder.setHeader(AwsHeaders.CHECKPOINTER, this.checkpointer); } - performSend(messageBuilder, records); + performSend(messageBuilder, record); if (CheckpointMode.record.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) { this.checkpointer.checkpoint(record.getSequenceNumber()); @@ -954,7 +973,14 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private void performSend(AbstractIntegrationMessageBuilder messageBuilder, Object rawRecord) { Message messageToSend = messageBuilder.build(); setAttributesIfNecessary(rawRecord, messageToSend); - sendMessage(messageToSend); + try { + sendMessage(messageToSend); + } + catch (Exception e) { + logger.info("Got an exception during sending a '" + messageToSend + "'" + + "\nfor the '" + rawRecord + "'.\n" + + "Consider to use 'errorChannel' flow for the compensation logic.", e); + } } @Override @@ -1015,7 +1041,13 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } else { if (shardConsumer.task != null) { - shardConsumer.task.run(); + try { + shardConsumer.task.run(); + } + catch (Exception e) { + logger.info("Got an exception " + e + " during [" + shardConsumer + "] task invocation.\n" + + "Process will be retried on the next iteration."); + } } } } diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java index ff5e2f4..ff8bb5f 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -30,7 +30,7 @@ import com.amazonaws.services.kinesis.model.Record; /** * An internal {@link Checkpointer} implementation based on - * provided {@link MetadataStore} and unikey {@code key} for shard. + * provided {@link MetadataStore} and {@code key} for shard. *

* The instances of this class is created by the {@link KinesisMessageDrivenChannelAdapter} * for each {@code ShardConsumer}. @@ -57,17 +57,18 @@ class ShardCheckpointer implements Checkpointer { } @Override - public void checkpoint() { - checkpoint(this.lastCheckpointValue); + public boolean checkpoint() { + return checkpoint(this.lastCheckpointValue); } @Override - public void checkpoint(String sequenceNumber) { + public boolean checkpoint(String sequenceNumber) { if (this.active) { String existingSequence = this.checkpointStore.get(this.key); if (existingSequence == null || - new BigInteger(existingSequence).compareTo(new BigInteger(sequenceNumber)) <= 0) { + new BigInteger(existingSequence).compareTo(new BigInteger(sequenceNumber)) < 0) { this.checkpointStore.put(this.key, sequenceNumber); + return true; } } else { @@ -75,6 +76,8 @@ class ShardCheckpointer implements Checkpointer { logger.info("The [" + this + "] has been closed. Checkpoints aren't accepted anymore."); } } + + return false; } List filterRecords(List records) { 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 bef79b9..2989c4c 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -42,6 +42,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; @@ -134,7 +136,19 @@ public class KinesisIntegrationTests { @Bean public PollableChannel errorChannel() { - return new QueueChannel(); + QueueChannel queueChannel = new QueueChannel(); + queueChannel.addInterceptor(new ChannelInterceptorAdapter() { + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + super.postSend(message, channel, sent); + + if (message instanceof ErrorMessage) { + throw (RuntimeException) ((ErrorMessage) message).getPayload(); + } + } + }); + return queueChannel; } } diff --git a/src/test/resources/log4j.properties b/src/test/resources/log4j.properties deleted file mode 100644 index d1ba7b6..0000000 --- a/src/test/resources/log4j.properties +++ /dev/null @@ -1,9 +0,0 @@ -log4j.rootCategory=WARN, stdout - -log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d %5p %c{1} [%t] : %m%n - -log4j.category.org.springframework=WARN -log4j.category.org.springframework.integration.aws=WARN -log4j.category.org.springframework.integration=WARN diff --git a/src/test/resources/log4j2-test.xml b/src/test/resources/log4j2-test.xml new file mode 100644 index 0000000..0149189 --- /dev/null +++ b/src/test/resources/log4j2-test.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + + +