You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2025/08/01 06:51:22 UTC
(flink) 02/03: [FLINK-38158][metrics] Allow to adjust transformation via SinkV2Provider
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6ad9a70e28993a5dd4e7620adb926017adc3709c
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon May 27 18:55:17 2024 +0200
[FLINK-38158][metrics] Allow to adjust transformation via SinkV2Provider
---
.../org/apache/flink/table/connector/sink/SinkV2Provider.java | 10 ++++++++++
.../table/planner/plan/nodes/exec/common/CommonExecSink.java | 8 ++++++--
2 files changed, 16 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/SinkV2Provider.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/SinkV2Provider.java
index 93ce6238950..7e2db2c06aa 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/SinkV2Provider.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/sink/SinkV2Provider.java
@@ -25,6 +25,8 @@ import org.apache.flink.table.data.RowData;
import javax.annotation.Nullable;
+import java.util.Collections;
+import java.util.Map;
import java.util.Optional;
/**
@@ -58,4 +60,12 @@ public interface SinkV2Provider extends DynamicTableSink.SinkRuntimeProvider, Pa
}
Sink<RowData> createSink();
+
+ /**
+ * @return additional variables that will be added to scope of the metrics reported from the
+ * {@link Sink} created by this {@link SinkV2Provider}.
+ */
+ default Map<String, String> getAdditionalMetricVariables() {
+ return Collections.emptyMap();
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
index d05152b7185..819cbc5c0b6 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
@@ -451,16 +451,20 @@ public abstract class CommonExecSink extends ExecNodeBase<Object>
sinkMeta,
sinkParallelism);
} else if (runtimeProvider instanceof SinkV2Provider) {
+ SinkV2Provider sinkV2Provider = (SinkV2Provider) runtimeProvider;
Transformation<RowData> sinkTransformation =
applyRowtimeTransformation(
inputTransform, rowtimeFieldIndex, sinkParallelism, config);
final DataStream<RowData> dataStream = new DataStream<>(env, sinkTransformation);
- final Transformation<?> transformation =
+ final Transformation<RowData> transformation =
DataStreamSink.forSink(
dataStream,
- ((SinkV2Provider) runtimeProvider).createSink(),
+ sinkV2Provider.createSink(),
CustomSinkOperatorUidHashes.DEFAULT)
.getTransformation();
+ sinkV2Provider
+ .getAdditionalMetricVariables()
+ .forEach(transformation::addMetricVariable);
transformation.setParallelism(sinkParallelism, sinkParallelismConfigured);
sinkMeta.fill(transformation);
return transformation;