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

GH-188: expose KplMessageHandler.flushDuration

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/188
Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/issues/154

Also fix `KinesisMessageDrivenChannelAdapter` to not ERROR unlock interrupt
when the `KinesisMessageDrivenChannelAdapter` is not active any more

**Cherry-pick to `2.3.x`**

# Conflicts:
#	src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
This commit is contained in:
Artem Bilan
2021-03-22 16:37:40 -04:00
parent b1017a8990
commit 4a40d404e5
2 changed files with 17 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1557,8 +1557,13 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
try {
lock.unlock();
}
catch (Exception e) {
logger.error("Error during unlocking: " + lock, e);
catch (Exception ex) {
if (KinesisMessageDrivenChannelAdapter.this.active) {
logger.error("Error during unlocking: " + lock, ex);
}
else {
logger.info("Error during unlocking: " + lock + " while adapter was inactive", ex);
}
}
finally {
iterator.remove();

10
src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java
View File

@@ -180,6 +180,16 @@ public class KplMessageHandler extends AbstractAwsMessageHandler<Void> implement
this.embeddedHeadersMapper = embeddedHeadersMapper;
}
/**
* Configure a {@link Duration} how often to call a {@link KinesisProducer#flush()}.
* @param flushDuration the {@link Duration} to periodic call of a {@link KinesisProducer#flush()}.
* @since 2.3.6
*/
public void setFlushDuration(Duration flushDuration) {
Assert.notNull(flushDuration, "'flushDuration' must not be null.");
this.flushDuration = flushDuration;
}
/**
* Unsupported operation. Use {@link #setEmbeddedHeadersMapper} instead.
* @param headerMapper is not used.