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

GH-221: Revise an at-least-once delivery

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/221

* Modify the logic of the `KinesisMessageDrivenChannelAdapter` to rewind
a shard iterator to the failed sequence for any errors.
A rewinding sequence is determined from extra properties in the `ShardCheckpointer`
* Remove `RequestShardForSequenceException` since more natural behavior to
react for any record processor error without end-user interaction
This commit is contained in:
abilan
2023-04-25 17:11:07 -04:00
parent 63b40836ce
commit bc9a0825fb
5 changed files with 72 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@@ -433,8 +433,9 @@ For example, users may want to fully read any parent shards before starting to r
}
```
Starting with _version 3.0_. the `RequestShardForSequenceException` can be used for flow control to request the shard iterator for specific sequence.
For example, when consumer has failed processing batch at specific record, throwing this exception with a sequence of that record will ensure at-least-once delivery since the shard iterator will move back to the requested record sequence.
Starting with _version 3.0_, any exception thrown from the record process may lead to shard iterator rewinding to the latest check-pointed sequence or the first one in the current failed batch.
This ensures an at-least-once delivery for possibly failed records.
If the latest checkpoint is equal to the highest sequence in the batch, then shard consumer continue with the next iterator.
Also, the `KclMessageDrivenChannelAdapter` is provided for performing streams consumption by [Kinesis Client Library][].
See its JavaDocs for more information.

86
src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
View File

@@ -65,6 +65,7 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.core.AttributeAccessor;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.log.LogMessage;
import org.springframework.core.serializer.support.DeserializingConverter;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.aws.event.KinesisShardEndedEvent;
@@ -1094,18 +1095,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
this.shardIterator = result.nextShardIterator();
}
}
catch (RequestShardForSequenceException requestShardForSequenceException) {
// Something wrong happened and not all records were processed.
// Must start from the provided sequence
KinesisShardOffset newOffset = new KinesisShardOffset(this.shardOffset);
newOffset.setSequenceNumber(requestShardForSequenceException.getSequenceNumber());
newOffset.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER);
GetShardIteratorRequest shardIteratorRequest = newOffset.toShardIteratorRequest();
this.shardIterator =
KinesisMessageDrivenChannelAdapter.this.amazonKinesis
.getShardIterator(shardIteratorRequest)
.join()
.shardIterator();
catch (Exception ex) {
rewindIteratorOnError(ex, result);
}
finally {
attributesHolder.remove();
@@ -1154,6 +1145,39 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
};
}
private void rewindIteratorOnError(Exception ex, GetRecordsResponse result) {
KinesisShardOffset newOffset = new KinesisShardOffset(this.shardOffset);
String lastCheckpoint = this.checkpointer.getLastCheckpointValue();
String highestSequence = this.checkpointer.getHighestSequence();
if (highestSequence.equals(lastCheckpoint)) {
logger.info(ex, "Record processor has thrown exception. " +
"Ignore since the highest sequence in batch was check-pointed.");
this.shardIterator = result.nextShardIterator();
return;
}
String newOffsetValue = lastCheckpoint;
if (lastCheckpoint != null) {
newOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
}
else {
newOffsetValue = this.checkpointer.getFirstSequenceInBatch();
newOffset.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER);
}
logger.info(ex,
LogMessage.format("Record processor has thrown exception. " +
"Rewind shard iterator %s sequence number: %s",
(lastCheckpoint != null ? "after" : "at"), newOffsetValue));
newOffset.setSequenceNumber(newOffsetValue);
GetShardIteratorRequest shardIteratorRequest = newOffset.toShardIteratorRequest();
this.shardIterator =
KinesisMessageDrivenChannelAdapter.this.amazonKinesis
.getShardIterator(shardIteratorRequest)
.join()
.shardIterator();
}
private void checkpointSwallowingProvisioningExceptions(String endingSequenceNumber) {
try {
this.checkpointer.checkpoint(endingSequenceNumber);
@@ -1205,6 +1229,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
private void processRecords(List<Record> records) {
logger.trace(() -> "Processing records: " + records + " for [" + ShardConsumer.this + "]");
this.checkpointer.setFirstSequenceInBatch(records.get(0).sequenceNumber());
this.checkpointer.setHighestSequence(records.get(records.size() - 1).sequenceNumber());
if (ListenerMode.record.equals(KinesisMessageDrivenChannelAdapter.this.listenerMode)) {
@@ -1313,42 +1338,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
Message<?> messageToSend = messageBuilder.build();
setAttributesIfNecessary(rawRecord, messageToSend);
try {
sendMessage(messageToSend);
}
catch (RequestShardForSequenceException requestShardForSequenceException) {
// Rethrow
throw requestShardForSequenceException;
}
catch (Exception ex) {
RequestShardForSequenceException requestShardForSequenceExceptionInCause =
findRequestShardForSequenceExceptionInCause(ex);
if (requestShardForSequenceExceptionInCause != null) {
throw requestShardForSequenceExceptionInCause;
}
else {
logger.info(ex, () ->
"Got an exception during sending a '"
+ messageToSend
+ "'"
+ "\nfor the '"
+ rawRecord
+ "'.\n"
+ "Consider to use 'errorChannel' flow for the compensation logic.");
}
}
}
@Nullable
private static RequestShardForSequenceException findRequestShardForSequenceExceptionInCause(Throwable ex) {
if (ex instanceof RequestShardForSequenceException requestShardForSequenceException) {
return requestShardForSequenceException;
}
Throwable cause = ex.getCause();
if (cause != null && cause != ex) {
return findRequestShardForSequenceExceptionInCause(cause);
}
return null;
sendMessage(messageToSend);
}
private void checkpointIfBatchMode() {

48
src/main/java/org/springframework/integration/aws/inbound/kinesis/RequestShardForSequenceException.java
View File

@@ -1,48 +0,0 @@
/*
* Copyright 2023 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.inbound.kinesis;
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
/**
* The flow control exception to notify the {@link KinesisMessageDrivenChannelAdapter}
* that specific shard iterator ({@link ShardIteratorType#AT_SEQUENCE_NUMBER})
* must be requested instead of checkpointing.
*
* @author Artem Bilan
*
* @since 3.0
*/
@SuppressWarnings("serial")
public class RequestShardForSequenceException extends RuntimeException {
private final String sequenceNumber;
public RequestShardForSequenceException(String sequenceNumber) {
this.sequenceNumber = sequenceNumber;
}
public RequestShardForSequenceException(String sequenceNumber, Throwable cause) {
super(cause);
this.sequenceNumber = sequenceNumber;
}
public String getSequenceNumber() {
return this.sequenceNumber;
}
}

35
src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2023 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.
@@ -23,6 +23,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.lang.Nullable;
/**
* An internal {@link Checkpointer} implementation based on provided {@link MetadataStore}
@@ -42,6 +43,10 @@ class ShardCheckpointer implements Checkpointer {
private final String key;
private volatile String firstSequenceInBatch;
private volatile String highestSequence;
private volatile String lastCheckpointValue;
private volatile boolean active = true;
@@ -53,7 +58,7 @@ class ShardCheckpointer implements Checkpointer {
@Override
public boolean checkpoint() {
return checkpoint(this.lastCheckpointValue);
return checkpoint(this.highestSequence);
}
@Override
@@ -66,7 +71,11 @@ class ShardCheckpointer implements Checkpointer {
return this.checkpointStore.replace(this.key, existingSequence, sequenceNumber);
}
else {
return this.checkpointStore.putIfAbsent(this.key, sequenceNumber) == null;
boolean stored = this.checkpointStore.putIfAbsent(this.key, sequenceNumber) == null;
if (stored) {
this.lastCheckpointValue = sequenceNumber;
}
return stored;
}
}
}
@@ -79,14 +88,30 @@ class ShardCheckpointer implements Checkpointer {
return false;
}
void setFirstSequenceInBatch(String firstSequenceInBatch) {
this.firstSequenceInBatch = firstSequenceInBatch;
}
@Nullable
String getFirstSequenceInBatch() {
return this.firstSequenceInBatch;
}
void setHighestSequence(String highestSequence) {
this.lastCheckpointValue = highestSequence;
this.highestSequence = highestSequence;
}
String getHighestSequence() {
return this.highestSequence;
}
@Nullable
String getCheckpoint() {
return this.checkpointStore.get(this.key);
this.lastCheckpointValue = this.checkpointStore.get(this.key);
return this.lastCheckpointValue;
}
@Nullable
String getLastCheckpointValue() {
return this.lastCheckpointValue;
}

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

@@ -36,7 +36,6 @@ import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.aws.LocalstackContainerTest;
import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter;
import org.springframework.integration.aws.inbound.kinesis.KinesisMessageHeaderErrorMessageStrategy;
import org.springframework.integration.aws.inbound.kinesis.RequestShardForSequenceException;
import org.springframework.integration.aws.outbound.KinesisMessageHandler;
import org.springframework.integration.aws.support.AwsHeaders;
import org.springframework.integration.channel.QueueChannel;
@@ -224,9 +223,7 @@ public class KinesisIntegrationTests implements LocalstackContainerTest {
@Override
public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
if (message instanceof ErrorMessage errorMessage && this.thrown.compareAndSet(false, true)) {
throw new RequestShardForSequenceException(
errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD, Record.class).sequenceNumber(),
errorMessage.getPayload());
throw (RuntimeException) errorMessage.getPayload();
}
}