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

GH-148: Check for periodic checkpoint mode properly

Browse Source 
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 213d793369)
This commit is contained in:
Dirk Bonhomme
2019-07-10 10:23:38 -04:00
committed by Artem Bilan
parent 60f0004afd
commit c4e288f7c8
1 changed files with 3 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;
}
}
/**