From c4e288f7c898ee69e690d57de04ebcf7f432d521 Mon Sep 17 00:00:00 2001 From: Dirk Bonhomme Date: Wed, 10 Jul 2019 10:23:38 -0400 Subject: [PATCH] GH-148: Check for periodic checkpoint mode properly Fixes https://github.com/spring-projects/spring-integration-aws/issues/148 Check for configured checkpoint mode before committing periodic checkpoint in KCL adapter (cherry picked from commit 213d7933694d2a70dc927041ccd7967e26345fb0) --- .../kinesis/KclMessageDrivenChannelAdapter.java | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java index 581d240..c2fa167 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java @@ -67,6 +67,7 @@ import com.amazonaws.services.kinesis.model.Record; * * @author Hervé Fortin * @author Artem Bilan + * @author Dirk Bonhomme * * @since 2.2.0 */ @@ -317,7 +318,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { finally { attributesHolder.remove(); // Checkpoint once every checkpoint interval. - if (System.currentTimeMillis() > nextCheckpointTimeInMillis) { + if (CheckpointMode.periodic.equals(KclMessageDrivenChannelAdapter.this.checkpointMode) && + System.currentTimeMillis() > nextCheckpointTimeInMillis) { checkpoint(checkpointer); this.nextCheckpointTimeInMillis = System.currentTimeMillis() + checkpointsInterval; } @@ -328,13 +330,6 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { if (CheckpointMode.batch.equals(KclMessageDrivenChannelAdapter.this.checkpointMode)) { checkpoint(checkpointer); } - else if (CheckpointMode.periodic.equals(KclMessageDrivenChannelAdapter.this.checkpointMode) && - System.currentTimeMillis() > nextCheckpointTimeInMillis) { - - checkpoint(checkpointer); - this.nextCheckpointTimeInMillis = System.currentTimeMillis() + checkpointsInterval; - } - } /**