GH-188: expose KplMessageHandler.flushDuration

Fixes #188
Fixes spring-cloud/spring-cloud-stream-binder-aws-kinesis#154

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

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

Author: artembilan

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

@@ -1535,7 +1535,12 @@ public void run() {
 						lock.unlock();
 					}
 					catch (Exception ex) {
-						logger.error(ex, () -> "Error during unlocking: " + lock);
+						if (KinesisMessageDrivenChannelAdapter.this.active) {
+							logger.error(ex, () -> "Error during unlocking: " + lock);
+						}
+						else {
+							logger.info(ex, () -> "Error during unlocking: " + lock + " while adapter was inactive");
+						}
 					}
 					finally {
 						iterator.remove();

src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java

@@ -180,6 +180,16 @@ public void setEmbeddedHeadersMapper(OutboundMessageMapper<byte[]> embeddedHeade
 		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.