Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/4871#discussion_r146467213
--- Diff:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisProducer.java
---
@@ -265,19 +259,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) {
--- End diff --
@bowenli86 I don't think that would happen. Records are added to the
producer only in `invoke`, which is guaranteed to not be executed concurrently
with `snapshotState`.
---
