update

apache · Oct 31, 2024 · a3fa78c · a3fa78c
1 parent 50dcf2c
commit a3fa78c
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 6 deletions.
diff --git a/...c/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusSink.java b/...c/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusSink.java
@@ -19,24 +19,25 @@
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.sink.SupportMultiTableSink;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
 import org.apache.seatunnel.connectors.seatunnel.http.config.HttpConfig;
 import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 
 public class PrometheusSink extends AbstractSimpleSink<SeaTunnelRow, Void>
         implements SupportMultiTableSink {
 
     protected final HttpParameter httpParameter = new HttpParameter();
-    protected SeaTunnelRowType seaTunnelRowType;
+    protected CatalogTable catalogTable;
     protected ReadonlyConfig pluginConfig;
 
-    public PrometheusSink(ReadonlyConfig pluginConfig, SeaTunnelRowType rowType) {
+    public PrometheusSink(ReadonlyConfig pluginConfig, CatalogTable catalogTable) {
         this.pluginConfig = pluginConfig;
         httpParameter.setUrl(pluginConfig.get(HttpConfig.URL));
         if (pluginConfig.getOptional(HttpConfig.HEADERS).isPresent()) {
@@ -45,7 +46,7 @@ public PrometheusSink(ReadonlyConfig pluginConfig, SeaTunnelRowType rowType) {
         if (pluginConfig.getOptional(HttpConfig.PARAMS).isPresent()) {
             httpParameter.setHeaders(pluginConfig.get(HttpConfig.PARAMS));
         }
-        this.seaTunnelRowType = rowType;
+        this.catalogTable = catalogTable;
 
         if (Objects.isNull(httpParameter.getHeaders())) {
             Map<String, String> headers = new HashMap<>();
@@ -67,6 +68,12 @@ public String getPluginName() {
 
     @Override
     public PrometheusWriter createWriter(SinkWriter.Context context) {
-        return new PrometheusWriter(seaTunnelRowType, httpParameter, pluginConfig);
+        return new PrometheusWriter(
+                catalogTable.getSeaTunnelRowType(), httpParameter, pluginConfig);
+    }
+
+    @Override
+    public Optional<CatalogTable> getWriteCatalogTable() {
+        return Optional.ofNullable(catalogTable);
     }
 }
diff --git a/...java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusSinkFactory.java b/...java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusSinkFactory.java
@@ -39,7 +39,7 @@ public TableSink createSink(TableSinkFactoryContext context) {
 
         ReadonlyConfig readonlyConfig = context.getOptions();
         CatalogTable catalogTable = context.getCatalogTable();
-        return () -> new PrometheusSink(readonlyConfig, catalogTable.getSeaTunnelRowType());
+        return () -> new PrometheusSink(readonlyConfig, catalogTable);
     }
 
     @Override