flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From pnowojski <...@git.apache.org>
Subject [GitHub] flink pull request #4871: [FLINK-7637] [kinesis] Fix at-least-once guarantee...
Date Mon, 23 Oct 2017 09:03:07 GMT
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`.


---

Mime
View raw message