[Feature][Zeta engine] Added the metrics information of table statistics in multi-table mode #6959

Author: hawk9821

seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/MultiTableMetricsIT.java

@@ -97,10 +97,18 @@ public void multiTableMetrics() {
                                     .body("jobStatus", equalTo("FINISHED"))
                                     .body("metrics.SourceReceivedCount", equalTo("50"))
                                     .body("metrics.SinkWriteCount", equalTo("50"))
-                                    .body("metrics.TableSourceReceivedCount.fake1", equalTo("20"))
-                                    .body("metrics.TableSourceReceivedCount.fake2", equalTo("30"))
-                                    .body("metrics.TableSinkWriteCount.fake1", equalTo("20"))
-                                    .body("metrics.TableSinkWriteCount.fake2", equalTo("30"));
+                                    .body(
+                                            "metrics.TableSourceReceivedCount.'fake.table1'",
+                                            equalTo("20"))
+                                    .body(
+                                            "metrics.TableSourceReceivedCount.'fake.public.table2'",
+                                            equalTo("30"))
+                                    .body(
+                                            "metrics.TableSinkWriteCount.'fake.table1'",
+                                            equalTo("20"))
+                                    .body(
+                                            "metrics.TableSinkWriteCount.'fake.public.table2'",
+                                            equalTo("30"));
                         });
     }
 

seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelClientTest.java

@@ -587,10 +587,11 @@ public void testGetMultiTableJobMetrics() {
 
             String jobMetrics = jobClient.getJobMetrics(jobId);
 
-            Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_COUNT + "#fake1"));
-            Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_COUNT + "#fake2"));
-            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake1"));
-            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake2"));
+            Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_COUNT + "#fake.table1"));
+            Assertions.assertTrue(
+                    jobMetrics.contains(SOURCE_RECEIVED_COUNT + "#fake.public.table2"));
+            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake.table1"));
+            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake.public.table2"));
 
             log.info("jobMetrics : {}", jobMetrics);
             JsonNode jobMetricsStr = new ObjectMapper().readTree(jobMetrics);

seatunnel-engine/seatunnel-engine-client/src/test/resources/batch_fake_multi_table_to_console.conf

@@ -29,6 +29,7 @@ source {
     result_table_name = "fake1"
     row.num = 20
     schema = {
+      table = "fake.table1"
       fields {
         name = "string"
         age = "int"
@@ -41,6 +42,7 @@ source {
     result_table_name = "fake2"
     row.num = 30
     schema = {
+      table = "fake.public.table2"
       fields {
         name = "string"
         age = "int"

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java

@@ -382,7 +382,7 @@ private Map<String, Object> getJobMetrics(String jobMetrics) {
                     .forEach(
                             metricName -> {
                                 String tableName =
-                                        TablePath.of(metricName.split("#")[1]).getTableName();
+                                        TablePath.of(metricName.split("#")[1]).getFullName();
                                 if (metricName.startsWith(SOURCE_RECEIVED_COUNT)) {
                                     tableSourceReceivedCountMap.put(
                                             tableName, jobMetricsStr.get(metricName));

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java

@@ -17,11 +17,11 @@
 
 package org.apache.seatunnel.engine.server.task;
 
-import lombok.extern.slf4j.Slf4j;
 import org.apache.seatunnel.api.common.metrics.Counter;
 import org.apache.seatunnel.api.common.metrics.Meter;
 import org.apache.seatunnel.api.common.metrics.MetricsContext;
 import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.event.SchemaChangeEvent;
 import org.apache.seatunnel.api.table.event.handler.DataTypeChangeEventDispatcher;
 import org.apache.seatunnel.api.table.event.handler.DataTypeChangeEventHandler;
@@ -35,10 +35,17 @@
 import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
 import org.apache.seatunnel.engine.server.task.flow.OneInputFlowLifeCycle;
 
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES;
@@ -124,8 +131,7 @@ public void collect(T row) {
                 sourceReceivedBytesPerSeconds.markEvent(size);
                 flowControlGate.audit((SeaTunnelRow) row);
                 if (StringUtils.isNotEmpty(tableId)) {
-                    String tableName =
-                            Optional.of(TablePath.of(tableId).getFullName()).orElse("null");
+                    String tableName = getFullName(TablePath.of(tableId));
                     Counter sourceTableCounter = sourceReceivedCountPerTable.get(tableName);
                     if (Objects.nonNull(sourceTableCounter)) {
                         sourceTableCounter.inc();
@@ -226,4 +232,12 @@ public void sendRecordToNext(Record<?> record) throws IOException {
             }
         }
     }
+
+    private String getFullName(TablePath tablePath) {
+        if (StringUtils.isBlank(tablePath.getTableName())) {
+            tablePath =
+                    TablePath.of(tablePath.getDatabaseName(), tablePath.getSchemaName(), "default");
+        }
+        return tablePath.getFullName();
+    }
 }

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java

@@ -22,6 +22,7 @@
 import org.apache.seatunnel.api.source.SourceSplit;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
 import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.core.starter.flowcontrol.FlowControlStrategy;
@@ -87,12 +88,7 @@ public void init() throws Exception {
                 tablePaths =
                         producedCatalogTables.stream()
                                 .map(CatalogTable::getTableId)
-                                .map(
-                                        tableIdentifier ->
-                                                TablePath.of(
-                                                        tableIdentifier.getDatabaseName(),
-                                                        tableIdentifier.getSchemaName(),
-                                                        tableIdentifier.getTableName()))
+                                .map(TableIdentifier::toTablePath)
                                 .collect(Collectors.toList());
             } catch (UnsupportedOperationException e) {
                 // TODO remove it when all connector use `getProducedCatalogTables`

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SinkFlowLifeCycle.java

@@ -17,8 +17,6 @@
 
 package org.apache.seatunnel.engine.server.task.flow;
 
-import com.hazelcast.cluster.Address;
-import lombok.extern.slf4j.Slf4j;
 import org.apache.seatunnel.api.common.metrics.Counter;
 import org.apache.seatunnel.api.common.metrics.Meter;
 import org.apache.seatunnel.api.common.metrics.MetricsContext;
@@ -28,6 +26,8 @@
 import org.apache.seatunnel.api.sink.SinkCommitter;
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.sink.SupportResourceShare;
+import org.apache.seatunnel.api.sink.multitablesink.MultiTableSink;
+import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.event.SchemaChangeEvent;
 import org.apache.seatunnel.api.table.type.Record;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -45,15 +45,22 @@
 import org.apache.seatunnel.engine.server.task.operation.sink.SinkRegisterOperation;
 import org.apache.seatunnel.engine.server.task.record.Barrier;
 
+import org.apache.commons.lang3.StringUtils;
+
+import com.hazelcast.cluster.Address;
+import lombok.extern.slf4j.Slf4j;
+
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;