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

Upgrade dependencies; prepare for release

Browse Source
This commit is contained in:
Artem Bilan
2019-12-26 14:14:30 -05:00
parent 3a64f35162
commit 112da0085c
3 changed files with 11 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -521,7 +521,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
*/
private void checkpoint(IRecordProcessorCheckpointer checkpointer, @Nullable Record record) {
if (logger.isInfoEnabled()) {
logger.info("Checkpointing shard " + shardId);
logger.info("Checkpointing shard " + this.shardId);
}
try {
if (record == null) {
@@ -564,9 +564,10 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
private void checkpointIfPeriodicMode(IRecordProcessorCheckpointer checkpointer, @Nullable Record record) {
if (CheckpointMode.periodic.equals(KclMessageDrivenChannelAdapter.this.checkpointMode)
&& System.currentTimeMillis() > nextCheckpointTimeInMillis) {
&& System.currentTimeMillis() > this.nextCheckpointTimeInMillis) {
checkpoint(checkpointer, record);
this.nextCheckpointTimeInMillis = System.currentTimeMillis() + checkpointsInterval;
this.nextCheckpointTimeInMillis =
System.currentTimeMillis() + KclMessageDrivenChannelAdapter.this.checkpointsInterval;
}
}

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

@@ -1110,14 +1110,15 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
private void checkpointIfPeriodicMode(@Nullable Record record) {
if (CheckpointMode.periodic.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)
&& System.currentTimeMillis() > nextCheckpointTimeInMillis) {
&& System.currentTimeMillis() > this.nextCheckpointTimeInMillis) {
if (record == null) {
this.checkpointer.checkpoint();
}
else {
this.checkpointer.checkpoint(record.getSequenceNumber());
}
this.nextCheckpointTimeInMillis = System.currentTimeMillis() + checkpointsInterval;
this.nextCheckpointTimeInMillis =
System.currentTimeMillis() + KinesisMessageDrivenChannelAdapter.this.checkpointsInterval;
}
}