Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/6021#discussion_r197065961
--- Diff:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisProducer.java
---
@@ -144,6 +163,17 @@ public void setFailOnError(boolean failOnError) {
this.failOnError = failOnError;
}
+ /**
+ * The {@link KinesisProducer} holds an unbounded queue internally. To
avoid memory
+ * problems under high loads, a limit can be employed above which the
internal queue
+ * will be flushed, thereby applying backpressure.
+ *
+ * @param queueLimit The maximum length of the internal queue before
backpressuring
+ */
+ public void setQueueLimit(int queueLimit) {
+ this.queueLimit = queueLimit;
--- End diff --
Will need argument checks on the given `queueLimit`.
---
