Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/4871#discussion_r146197645
--- Diff:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisProducer.java
---
@@ -265,19 +257,86 @@ public void close() throws Exception {
if (kp != null) {
LOG.info("Flushing outstanding {} records",
kp.getOutstandingRecordsCount());
// try to flush all outstanding records
- while (kp.getOutstandingRecordsCount() > 0) {
- kp.flush();
- try {
- Thread.sleep(500);
- } catch (InterruptedException e) {
- LOG.warn("Flushing was interrupted.");
- // stop the blocking flushing and
destroy producer immediately
- break;
- }
- }
+ flushSync(kp);
--- End diff --
why do we need this `kp` local variable? Without it, there would be no need
to passe `KinesisProducer` as a param to `flushSync`, because `flushSync()`
could just use `this.producer`.
---
