GH-84: Kinesis Inbound: errors immunity

Fixes spring-projects/spring-integration-aws#84
Fixes spring-cloud/spring-cloud-stream-binder-aws-kinesis#38
Fixes spring-cloud/spring-cloud-stream-binder-aws-kinesis#36

Even if AWS Client has some reconnect and retry mechanism, it can be
exhausted and no connection error is rethrown to the `KinesisMessageDrivenChannelAdapter`
anyway.

On the other hand the error can be thrown from the record processor -
the flow on the `outputChannel`.

* log the exception around AWS Client calls and let background process
to restore/retry
* log the exception around message to send to let the processor to move
to the next record or perform the next task
* null the current `task` in the `ShardConsumer` in the `finally` block
to avoid hanging the thread without ability to moving to some other state
without end-user interaction
* When perform the `batch` checkpoint, check the result and if it is
negative, consider such a situation as processed and skip records from
sending downstream
* Upgrade dependencies
* Use Log4J2 for tests logging

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'

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

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<Shard> 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<ShardConsumer> 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<Record> records = result.getRecords();
 
@@ -895,8 +907,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i
 			ShardConsumer.this.state = ConsumerState.SLEEP;
 		}
 
-		private void processRecords(List<Record> records) {
-			records = this.checkpointer.filterRecords(records);
+		private void processRecords(List<Record> recordsToProcess) {
+			List<Record> 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.");
+							}
 						}
 					}
 				}

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.
  * <p>
  * 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<Record> filterRecords(List<Record> records) {

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;
 		}
 
 	}

src/test/resources/log4j.properties

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

src/test/resources/log4j2-test.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+	<Appenders>
+		<Console name="STDOUT" target="SYSTEM_OUT">
+			<PatternLayout pattern="%d %p [%t] [%c] - %m%n" />
+		</Console>
+	</Appenders>
+	<Loggers>
+		<Logger name="org.springframework" level="warn"/>
+		<Logger name="org.springframework.integration" level="warn"/>
+		<Logger name="org.springframework.integration.aws" level="info"/>
+		<Root level="warn">
+			<AppenderRef ref="STDOUT" />
+		</Root>
+	</Loggers>
+</Configuration>