Hisoka-X commented on code in PR #7212:
URL: https://github.com/apache/seatunnel/pull/7212#discussion_r1686116520
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java:
##########
@@ -77,17 +88,26 @@ public SeaTunnelSourceCollector(
List<OneInputFlowLifeCycle<Record<?>>> outputs,
MetricsContext metricsContext,
FlowControlStrategy flowControlStrategy,
- SeaTunnelDataType rowType) {
+ SeaTunnelDataType rowType,
+ List<TablePath> tablePaths) {
this.checkpointLock = checkpointLock;
this.outputs = outputs;
this.rowType = rowType;
+ this.metricsContext = metricsContext;
if (rowType instanceof MultipleRowType) {
((MultipleRowType) rowType)
.iterator()
- .forEachRemaining(
- type -> {
- this.rowTypeMap.put(type.getKey(),
type.getValue());
- });
+ .forEachRemaining(type ->
this.rowTypeMap.put(type.getKey(), type.getValue()));
+ }
+ if (CollectionUtils.isNotEmpty(tablePaths)) {
+ tablePaths.forEach(
+ tablePath ->
+ sourceReceivedCountPerTable.put(
+ tablePath.getTableName(),
Review Comment:
```suggestion
tablePath.getFullName(),
```
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java:
##########
@@ -100,21 +120,32 @@ public SeaTunnelSourceCollector(
public void collect(T row) {
try {
if (row instanceof SeaTunnelRow) {
+ String tableId = ((SeaTunnelRow) row).getTableId();
int size;
if (rowType instanceof SeaTunnelRowType) {
size = ((SeaTunnelRow)
row).getBytesSize((SeaTunnelRowType) rowType);
} else if (rowType instanceof MultipleRowType) {
- size =
- ((SeaTunnelRow) row)
- .getBytesSize(
- rowTypeMap.get(((SeaTunnelRow)
row).getTableId()));
+ size = ((SeaTunnelRow)
row).getBytesSize(rowTypeMap.get(tableId));
} else {
throw new SeaTunnelEngineException(
"Unsupported row type: " +
rowType.getClass().getName());
}
sourceReceivedBytes.inc(size);
sourceReceivedBytesPerSeconds.markEvent(size);
flowControlGate.audit((SeaTunnelRow) row);
+ if (StringUtils.isNotEmpty(tableId)) {
+ String tableName =
+
Optional.of(TablePath.of(tableId).getTableName()).orElse("null");
Review Comment:
```suggestion
Optional.of(TablePath.of(tableId).getFullName()).orElse("null");
```
##########
.github/workflows/backend.yml:
##########
@@ -313,7 +313,7 @@ jobs:
- name: run updated modules integration test (part-1)
if: needs.changes.outputs.api == 'false' &&
needs.changes.outputs.it-modules != ''
run: |
- sub_modules=`python
tools/update_modules_check/update_modules_check.py sub_update_it_module
${{needs.changes.outputs.it-modules}} 8 0`
+ sub_modules=`python
tools/update_modules_check/update_modules_check.py sub_update_it_module
${{needs.changes.outputs.it-modules}} 10 0`
Review Comment:
Please revert ci config change, the dev branch CI work fine at now.
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java:
##########
@@ -77,17 +88,26 @@ public SeaTunnelSourceCollector(
List<OneInputFlowLifeCycle<Record<?>>> outputs,
MetricsContext metricsContext,
FlowControlStrategy flowControlStrategy,
- SeaTunnelDataType rowType) {
+ SeaTunnelDataType rowType,
+ List<TablePath> tablePaths) {
this.checkpointLock = checkpointLock;
this.outputs = outputs;
this.rowType = rowType;
+ this.metricsContext = metricsContext;
if (rowType instanceof MultipleRowType) {
((MultipleRowType) rowType)
.iterator()
- .forEachRemaining(
- type -> {
- this.rowTypeMap.put(type.getKey(),
type.getValue());
- });
+ .forEachRemaining(type ->
this.rowTypeMap.put(type.getKey(), type.getValue()));
+ }
+ if (CollectionUtils.isNotEmpty(tablePaths)) {
+ tablePaths.forEach(
+ tablePath ->
+ sourceReceivedCountPerTable.put(
+ tablePath.getTableName(),
+ metricsContext.counter(
+ SOURCE_RECEIVED_COUNT
+ + "#"
+ +
tablePath.getTableName())));
Review Comment:
```suggestion
+
tablePath.getFullName())));
```
##########
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java:
##########
@@ -76,10 +81,25 @@ public void init() throws Exception {
+ startFlowLifeCycle.getClass().getName());
} else {
SeaTunnelDataType sourceProducedType;
+ List<TablePath> tablePaths = new ArrayList<>();
try {
List<CatalogTable> producedCatalogTables =
sourceFlow.getAction().getSource().getProducedCatalogTables();
sourceProducedType =
CatalogTableUtil.convertToDataType(producedCatalogTables);
+ tablePaths =
+ producedCatalogTables.stream()
+ .map(CatalogTable::getTableId)
+ .filter(
+ tableIdentifier ->
+ StringUtils.isNotBlank(
+
tableIdentifier.getTableName()))
Review Comment:
```suggestion
tableIdentifier.getFullName()))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
