[Bug] [Seatunnel flink starter] null row causes NPE

[litiliu]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

I'm running seatunnel 2.3.9 but got NPE. It seems the null row was not filtered before processing by the downstream operator.

SeaTunnel Version

2.3.9

SeaTunnel Config

{
  "env": {
    "jobMode": "batch",
    "parallelism": 1
  },
  "source": [
    {
      "plugin_name": "FakeSource",
      "plugin_output": "fake",
      "row.num": 5,
      "schema": {
        "fields": {
          "name": "string",
          "age": "int",
          "card": "int"
        }
      }
    }
  ],
  "transform": [
    {
      "plugin_name": "FilterRowKind",
      "plugin_input": "fake",
      "plugin_output": "fake1",
      "exclude_kinds": ["INSERT"]
    },
    {
      "plugin_name": "Copy",
      "plugin_input": "fake1",
      "plugin_output": "fake2",
      "fields": {
        "name1": "name",
        "age1": "age",
        "card1": "card"
      }
    }
  ],
  "sink": [
    {
      "plugin_name": "Console",
      "plugin_input": "fake2"
    }
  ]
}

Running Command

src/main/java/org/apache/seatunnel/example/flink/v2/SeaTunnelApiExample.java

Error Exception

`
Caused by: java.lang.NullPointerException
	at org.apache.seatunnel.api.table.type.SeaTunnelRowAccessor.getField(SeaTunnelRowAccessor.java:41)
	at org.apache.seatunnel.transform.copy.CopyFieldTransform.getOutputFieldValues(CopyFieldTransform.java:122)
	at org.apache.seatunnel.transform.common.MultipleFieldOutputTransform.transformRow(MultipleFieldOutputTransform.java:56)
	at org.apache.seatunnel.transform.common.MultipleFieldOutputTransform.transformRow(MultipleFieldOutputTransform.java:36)
	at org.apache.seatunnel.transform.common.AbstractSeaTunnelTransform.transform(AbstractSeaTunnelTransform.java:80)
	at org.apache.seatunnel.transform.common.AbstractCatalogSupportMapTransform.map(AbstractCatalogSupportMapTransform.java:42)
	at org.apache.seatunnel.transform.common.AbstractCatalogSupportMapTransform.map(AbstractCatalogSupportMapTransform.java:27)
	at org.apache.seatunnel.transform.common.AbstractMultiCatalogMapTransform.map(AbstractMultiCatalogMapTransform.java:40)
	at org.apache.seatunnel.transform.common.AbstractMultiCatalogMapTransform.map(AbstractMultiCatalogMapTransform.java:28)
	at org.apache.seatunnel.core.starter.flink.execution.TransformExecuteProcessor.lambda$flinkTransform$5d1c7633$1(TransformExecuteProcessor.java:159)
	at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:38)

`

Zeta or Flink or Spark Version

flink 1.18.1

Java or Scala Version

java 11

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[litiliu]

The null row generated in the FilterRowKind plugin was directly forwarded to the downstream plugins, causing the downstream plugin NPE.