SEPURI-SAI-KRISHNA opened a new issue, #13045:
URL: https://github.com/apache/seatunnel/issues/13045
### Search before asking
- [X] I had searched in the issues and found no similar issues.
### What happened
`DruidWriter.write` encodes each row as a line of inline CSV, but leaves a
null field out instead of writing an empty value:
```java
for (int i = 0; i < element.getArity(); i++) {
final Object v = element.getField(i);
if (v != null) {
joiner.add(v.toString());
}
}
// timestamp column is a required field to add in Druid.
joiner.add(String.valueOf(processTime));
```
A row with a null comes out one field short, so every later value moves one
column to the left, including the processing time the writer appends as the
`timestamp` column. The indexing task declares its CSV columns as the row's
field names followed by `timestamp` (`provideDruidIOConfig`).
Measured with the real `DruidWriter` sending to a local HTTP endpoint, for
columns `id, name, amount` and the row `(2, null, 20)`, the submitted line is:
```
2,20,1793180345605
```
Read with Druid 24.0.1's own CSV reader (the version `connector-druid` pins)
and the task's columns, that line fails:
```
ParseException: Timestamp[null] is unparseable! Event: {id=2, name=20,
amount=1793180345605}
```
`name` received the amount and `amount` received the timestamp, which leaves
`timestamp` empty. `provideInputJSONString` removes `tuningConfig` from the
task, so Druid's defaults apply: `maxParseExceptions` is `Integer.MAX_VALUE`
and `logParseExceptions` is `false` (`TuningConfig` in druid-server 24.0.1).
The row is dropped without a log line, and the indexing task and the SeaTunnel
job both succeed.
`DruidIT` does not catch this: its rows contain no null value.
### What you expected to happen
A null field is written as an empty CSV value, so it keeps its column and
the row is ingested.
### SeaTunnel Version
dev, verified at commit `f7a575432`.
### SeaTunnel Config
```conf
env {
parallelism = 1
job.mode = "BATCH"
}
source {
FakeSource {
schema = {
fields {
id = int
name = string
amount = bigint
}
}
rows = [
{
kind = INSERT
fields = [1, "a", 10]
},
{
kind = INSERT
fields = [2, null, 20]
}
]
}
}
sink {
Druid {
coordinatorUrl = "localhost:8888"
datasource = "null_field_test"
}
}
```
### Running Command
```shell
./bin/seatunnel.sh --config config/fake_to_druid.conf -e local
```
### Error Exception
```log
None. The job succeeds. Druid's CSV parser rejects the row with the null,
and under the default limits the task discards it.
```
### Zeta or Flink or Spark Version
Not engine specific. The line is built in the sink writer.
### Java or Scala Version
Not version specific. Measured on JDK 17.
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's Code of Conduct
--
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]
