You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/09/07 10:18:26 UTC
[flink] 03/11: [FLINK-24131][datastream] Recommit recovered
transactions as quickly as possible.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b6da3342a15457ee4934ce1501c690c45ea80300
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sat Sep 4 22:35:41 2021 +0200
[FLINK-24131][datastream] Recommit recovered transactions as quickly as possible.
---
.../flink/streaming/runtime/operators/sink/CommitterOperator.java | 2 ++
.../org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java | 1 +
2 files changed, 3 insertions(+)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
index 125ec06..f72f1de 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -72,6 +72,8 @@ class CommitterOperator<InputT, OutputT> extends AbstractStreamOperator<byte[]>
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
committerHandler.initializeState(context);
+ // try to re-commit recovered transactions as quickly as possible
+ commitRetrier.retryWithDelay();
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
index 015e90c..dec4036 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
@@ -149,6 +149,7 @@ class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<b
checkpointId.isPresent() ? checkpointId.getAsLong() : null),
sinkWriterStateHandler.initializeState(context));
committerHandler.initializeState(context);
+ commitRetrier.retryWithDelay();
}
@Override