throw exception when bug happened

happyboy1024 · Oct 25, 2024 · 33710c2 · 33710c2
1 parent c9026ad
commit 33710c2
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/.../src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisSourceReader.java b/.../src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisSourceReader.java
@@ -23,6 +23,8 @@
 import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.connectors.doris.config.DorisSourceConfig;
+import org.apache.seatunnel.connectors.doris.exception.DorisConnectorErrorCode;
+import org.apache.seatunnel.connectors.doris.exception.DorisConnectorException;
 import org.apache.seatunnel.connectors.doris.rest.PartitionDefinition;
 import org.apache.seatunnel.connectors.doris.source.DorisSourceTable;
 import org.apache.seatunnel.connectors.doris.source.split.DorisSourceSplit;
@@ -78,7 +80,11 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
                 DorisSourceTable dorisSourceTable =
                         tables.get(TablePath.of(partition.getDatabase(), partition.getTable()));
                 if (dorisSourceTable == null) {
-                    return;
+                    throw new DorisConnectorException(
+                            DorisConnectorErrorCode.SHOULD_NEVER_HAPPEN,
+                            String.format(
+                                    "the table '%s.%s' cannot be found in table_list of job configuration.",
+                                    partition.getDatabase(), partition.getTable()));
                 }
                 valueReader = new DorisValueReader(partition, dorisSourceConfig, dorisSourceTable);
                 while (valueReader.hasNext()) {