flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tzulitai <...@git.apache.org>
Subject [GitHub] flink pull request #4871: [FLINK-7637] [kinesis] Fix at-least-once guarantee...
Date Tue, 24 Oct 2017 06:53:21 GMT
Github user tzulitai commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4871#discussion_r146466759
  
    --- 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);
    +
     			LOG.info("Flushing done. Destroying producer instance.");
     			kp.destroy();
     		}
    +
    +		// make sure we propagate pending errors
    +		checkAndPropagateAsyncError();
    +	}
    +
    +	@Override
    +	public void initializeState(FunctionInitializationContext context) throws Exception
{
    +		// nothing to do
    +	}
    +
    +	@Override
    +	public void snapshotState(FunctionSnapshotContext context) throws Exception {
    +		// check for asynchronous errors and fail the checkpoint if necessary
    +		checkAndPropagateAsyncError();
    +
    +		flushSync(producer);
    +		if (producer.getOutstandingRecordsCount() > 0) {
    +			throw new IllegalStateException(
    +				"Number of outstanding records must be zero at this point: " + producer.getOutstandingRecordsCount());
    +		}
    +
    +		// if the flushed requests has errors, we should propagate it also and fail the checkpoint
    +		checkAndPropagateAsyncError();
    +	}
    +
    +	// --------------------------- Utilities ---------------------------
    +
    +	/**
    +	 * Creates a {@link KinesisProducer}.
    +	 * Exposed so that tests can inject mock producers easily.
    +	 */
    +	@VisibleForTesting
    +	protected KinesisProducer getKinesisProducer(KinesisProducerConfiguration producerConfig)
{
    +		return new KinesisProducer(producerConfig);
    +	}
    +
    +	/**
    +	 * Check if there are any asynchronous exceptions. If so, rethrow the exception.
    +	 */
    +	private void checkAndPropagateAsyncError() throws Exception {
    --- End diff --
    
    yes, only a refactoring of the code to a separate method.


---

Mime
View raw message