Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/4871#discussion_r146198571
--- 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 --
I am assuming that during this moving - copy/pasting - there were no
changes in the code? (Btw, that's another reason why having refactors in
separate commits makes reviewing easier ;) )
---
