Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/4871#discussion_r146466699
--- 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 --
Makes sense, will change.
---
