SEPURI-SAI-KRISHNA opened a new pull request, #12953:
URL: https://github.com/apache/seatunnel/pull/12953
Yes, glad to be contributing to Apache SeaTunnel.
Closes #12952.
### Answers to the three questions before writing code
- **Issue, expected behaviour and acceptance criteria**: #12952. A Kafka
tombstone, whose value is `null`, must produce no row and let the offset
advance, rather than failing the source. Accepted as done when a null value
yields no row through both the `byte[]` and the `Collector` entry points, a
zero length payload still decodes to a message of protobuf defaults, and both
are pinned by tests that fail if the guard is removed.
- **Modules that change, and what must stay compatible**:
`seatunnel-format-protobuf` only, plus its two format documents. The wire
format, the option names, `ProtobufToRowConverter`, the descriptor compilation
and the serialization side are untouched. The `DeserializationSchema` contract
is unchanged: it already defines a `null` return as nothing to collect.
- **Required validation**: the module unit suite, a mutation check proving
the new tests fail without the fix, `spotless:check`, and `./mvnw -DskipTests
verify`. Results are at the end.
### Duplicate search
Searched open and closed issues and pull requests before implementing and
again immediately before opening this PR, for `protobuf tombstone`, `protobuf
null`, `protobuf NullPointerException`, `ProtobufDeserializationSchema` and
`DynamicMessage`. No duplicate exists. The only `tombstone` match is my own
#12927, which fixes the same defect for `avro` and names protobuf as a
follow-up.
Three open PRs touch `seatunnel-format-protobuf` and none overlaps this
change:
- **#12905** also edits `ProtobufDeserializationSchema.java`, at
`convertToRow` (`tablePath.toString()` to `tablePath.get().toString()`). That
is a different method about fifteen lines below the one changed here. I
verified with `git merge-tree` that this branch and #12905's head merge with no
conflict.
- **#12660** edits `ProtobufFieldResolver`, the two converters and the error
codes.
- **#12902** edits `CompileDescriptor`.
### What problem does this PR solve
`ProtobufDeserializationSchema.deserialize(byte[])` handed the value
straight to `DynamicMessage.parseFrom`:
```java
public SeaTunnelRow deserialize(byte[] message) throws IOException {
Descriptors.Descriptor descriptor = this.converter.getDescriptor();
DynamicMessage dynamicMessage = DynamicMessage.parseFrom(descriptor,
message);
return convertToRow(dynamicMessage);
}
```
A tombstone carries a `null` value, so the job failed with a
`NullPointerException` raised inside the protobuf runtime. Every other text
oriented format the Kafka source supports already passes over a tombstone.
Protobuf was the last one with no guard.
### Measured, not assumed
Against `protobuf-java` 3.25.3, the version this module declares, with a
descriptor for `message Person { string name = 1; int32 age = 2; }`:
```
null -> java.lang.NullPointerException: Cannot read the array length
because "data" is null
at
com.google.protobuf.AbstractMessageLite$Builder.mergeFrom(AbstractMessageLite.java:191)
at
com.google.protobuf.AbstractMessage$Builder.mergeFrom(AbstractMessage.java:505)
at
com.google.protobuf.DynamicMessage.parseFrom(DynamicMessage.java:95)
empty[0] -> returned a message, fields: name="" age=0 serializedSize=0
```
That second line is why this guard is null only rather than null or empty.
Protobuf defines a zero length payload as a valid message whose fields all hold
their defaults, so treating it as a tombstone would have broken correct
protobuf behaviour. `avro`, where a zero length buffer is not a valid datum, is
the opposite case, and that is why #12927 guards both there and this guards one
here. The resulting shape matches `json` and `maxwell_json`, which guard a null
value only.
### The fix
A null check at the top of `deserialize(byte[])` returning `null`, which
`DeserializationSchema.deserialize(byte[], Collector)` already treats as
nothing to collect:
```java
default void deserialize(byte[] message, Collector<T> out) throws
IOException {
T deserialize = deserialize(message);
if (deserialize != null) {
out.collect(deserialize);
}
}
```
### Every caller was audited, and none breaks on a null return
This is the one risk worth stating explicitly, because the method's return
value changes shape. There are exactly two production consumers of this class:
- **Kafka**, `KafkaSourceConfig:446`, used whenever
`strip_schema_registry_header` keeps its default. It reaches the format through
`KafkaRecordEmitter:94`, which calls the `Collector` overload above, so a null
return collects nothing and the offset still advances at
`KafkaRecordEmitter:116`.
- **RabbitMQ**, `RabbitmqSourceReader:134`, which calls the `byte[]`
overload directly and already null checks the result:
```java
SeaTunnelRow row = schema.deserialize(delivery.getBody());
if (row != null) {
row.setTableId(exactTableId);
output.collect(row);
}
```
RabbitMQ has to handle this already, because `JsonDeserializationSchema`
returns null for a null message on the same code path.
`SchemaRegistryAwareProtobufDeserializationSchema` is the third caller and
simply delegates, so it inherits the behaviour.
The `deserialize(InputStream)` overload is deliberately left alone. Its only
caller is `SchemaRegistryAwareProtobufDeserializationSchema:187`, which always
wraps a non null byte range in a `ByteArrayInputStream`, so it cannot receive
null.
### One existing assertion changes, and why
`SchemaRegistryAwareProtobufDeserializationSchemaTest` pinned the crash:
```java
@Test
void throwsWhenMessageIsNull() {
Assertions.assertThrows(NullPointerException.class, () ->
schema.deserialize(null));
}
```
I want to be direct about this rather than let it be discovered in review.
It became `returnsNoRowWhenMessageIsNull`, asserting `assertNull`. The original
was not a designed contract:
- it arrived with #12039, `[Fix][Format][Protobuf] Support empty Schema
Registry payloads`, where it replaced an older `testDeserializeEmptyMessage`.
That PR's subject was zero length payloads, and this assertion recorded what
the code happened to do with a null one.
- no option, document or error code describes a null value as a protobuf
format error. The exception came from inside the protobuf runtime, not from any
SeaTunnel check.
- every other format reached the opposite conclusion.
No other assertion in the repository pins this behaviour; I checked for
`NullPointerException` across every protobuf test.
### Tests
`ProtobufDeserializationSchemaTest` is new. The class had no test of its
own, and per the guidance on not adding test classes casually I looked for an
existing home first: `ProtobufConverterTest` covers the converters and
`SchemaRegistryAwareProtobufDeserializationSchemaTest` covers the registry
aware wrapper, so neither is the right owner for the plain schema. It is also
the entry point Kafka uses by default, which makes the gap worth closing.
Three cases, and a mutation check with the guard removed:
| test | with the fix | guard removed |
| --- | --- | --- |
| `deserializeReturnsNoRowForATombstone` | passes | `NullPointer` |
| `collectorReceivesNothingForATombstone` | passes | `NullPointer` |
| `anEmptyPayloadRemainsAValidMessageOfDefaults` | passes | **still passes**
|
| `returnsNoRowWhenMessageIsNull` (updated) | passes | `NullPointer` |
The empty payload case staying green with the guard removed is the point of
including it: it pins the boundary independently instead of riding on the fix.
### Documentation
`docs/en/connectors/formats/protobuf.md` and
`docs/zh/connectors/formats/protobuf.md` gain a matching `Tombstone and empty
messages` section, describing both the tombstone behaviour and the empty
payload behaviour that deliberately did not change. Neither file previously
said anything about null or empty values. The English file also gains the
trailing newline it was missing, which is the one unrelated line in the diff.
No `incompatible-changes.md` entry: the previous behaviour was a crash, not
a contract users could depend on.
### Relationship to #12927
#12927 adds a tombstone section to the Kafka source documents which
currently states that `protobuf` has no such guard
(`docs/en/connectors/source/Kafka.md:625` and
`docs/zh/connectors/source/Kafka.md:613`). Those lines become stale when this
merges. They are not in this PR's base, so they cannot be corrected here.
Whichever lands second should carry the correction, and I will make it in
whichever PR that is.
### No E2E in this iteration
`connector-kafka-e2e` has protobuf coverage, and the guidance to update an
existing IT applies. I have left it out of this iteration for one concrete
reason: #12660 is actively editing both `protobuf/fake_to_kafka_protobuf.conf`
and `protobuf/kafka_protobuf_to_assert.conf`, and adding a tombstone case now
would put this PR in conflict with it over files that have nothing to do with
this defect. The unit coverage exercises the real entry point including the
`Collector` path. Happy to add the E2E if a reviewer would rather have it now,
or as a follow-up once #12660 lands.
### Validation
All on JDK 17.
- `./mvnw test -pl seatunnel-formats/seatunnel-format-protobuf`: **22 tests,
0 failures, 0 errors, 0 skipped**
- `ProtobufDeserializationSchemaTest` 3,
`SchemaRegistryAwareProtobufDeserializationSchemaTest` 18,
`ProtobufConverterTest` 1
- mutation check as tabulated above, then the guard restored and the suite
re-run green
- `./mvnw spotless:check -pl seatunnel-formats/seatunnel-format-protobuf`:
clean
- `./mvnw -DskipTests verify -pl seatunnel-formats/seatunnel-format-protobuf
-am`: clean
- `git merge-tree` against current `dev` and against #12905's head: no
conflict
--
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]
