You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "arvindKandpal-ksolves (via GitHub)" <gi...@apache.org> on 2026/05/20 05:39:12 UTC
[PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
arvindKandpal-ksolves opened a new pull request, #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124
## Purpose of the change
This PR fixes [FLINK-36235](https://issues.apache.org/jira/browse/FLINK-36235).
Currently, when a message fails to deserialize and returns `null`, the `PulsarDeserializationSchemaWrapper` emits this `null` value downstream. This violates Flink's `DeserializationSchema` contract (where a `null` return value means "drop this record") and can cause `NullPointerException`s in downstream operators. This PR adds a simple null-check guard to safely drop these corrupted or null records.
## Brief change log
- Added an `if (instance != null)` check before emitting the record in `PulsarDeserializationSchemaWrapper#deserialize`.
- Added a regression test `wrapperDropsNullDeserializedRecord` and a `CountingCollector` helper class in `PulsarDeserializationSchemaTest` to verify that null records are correctly dropped without errors.
## Verifying this change
This change added tests and can be verified as follows:
- Added unit tests in `PulsarDeserializationSchemaTest.java` to explicitly test and verify the filtering behavior when the inner deserializer returns `null`.
- Verified locally using `mvn -pl flink-connector-pulsar test`.
## Significant changes
- [ ] Dependencies have been added or upgraded
- [ ] Public API has been changed (Public API is any class annotated with `@Public(Evolving)`)
- [ ] Serializers have been changed
- [ ] New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124#issuecomment-4494871235
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
Posted by "arvindKandpal-ksolves (via GitHub)" <gi...@apache.org>.
arvindKandpal-ksolves commented on PR #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124#issuecomment-4532275389
@featzhang Thanks for the detailed review! I have addressed all your feedback in the latest commit. I added the null checks and DEBUG logging (with message IDs) across all four wrapper classes (PulsarDeserializationSchemaWrapper, PulsarSchemaWrapper, PulsarTypeInformationWrapper, and GenericRecordDeserializationSchema). I also replaced the custom test collector with Flink's native ListCollector and updated the tests to cover the new behavior. Please let me know if it looks good to go!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
Posted by "arvindKandpal-ksolves (via GitHub)" <gi...@apache.org>.
arvindKandpal-ksolves commented on PR #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124#issuecomment-4729167397
Hi Team , Can any one review this pr for futher changes ?
cc @featzhang @davidradl
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
Posted by "arvindKandpal-ksolves (via GitHub)" <gi...@apache.org>.
arvindKandpal-ksolves commented on PR #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124#issuecomment-4520530952
Hi @featzhang , Can you review this Patch ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
Posted by "featzhang (via GitHub)" <gi...@apache.org>.
featzhang commented on PR #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124#issuecomment-4527998590
The change itself is correct and matches the default `deserialize(byte[], Collector)` implementation in `DeserializationSchema` (flink-core), which drops `null` returns by contract. Two things worth folding into this PR or a follow-up:
1. The same pattern exists in three other wrappers in the same package and is not addressed here:
- `PulsarSchemaWrapper.java:68`
- `PulsarTypeInformationWrapper.java:61`
- `GenericRecordDeserializationSchema.java:41`
They share the exact issue (`out.collect(instance)` without a null check). Since the JIRA describes a contract bug, fixing only one wrapper leaves the same footgun for the other code paths.
2. Behavior change for the table connector is worth calling out in the PR description. In non-upsert mode, a null value previously reached `PulsarRowDataConverter#emitRow` and threw `DeserializationException("Invalid null value received in non-upsert mode...")`. After this PR, the wrapper drops it silently. The new behavior is fine and aligns with Kafka, but it should not be hidden in the changelog — users currently relying on the exception to detect bad data will lose records silently. A `LOG.debug` on the dropped record would help operability.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
Posted by "featzhang (via GitHub)" <gi...@apache.org>.
featzhang commented on code in PR #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124#discussion_r3294214764
##########
flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java:
##########
@@ -373,6 +399,22 @@ private <T> Message<byte[]> getMessage(
return MessageImpl.create(metadata, payload, Schema.BYTES, "");
}
+ /** Collector that records every {@link #collect} invocation, including nulls. */
+ private static class CountingCollector<T> implements Collector<T> {
Review Comment:
Nit: `org.apache.flink.api.common.functions.util.ListCollector` would do the same job (collect into a `List<T>`) without introducing a new test helper. Not blocking.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36235][Stream] Ignore emitting null rowData when deserialized message fails [flink-connector-pulsar]
Posted by "arvindKandpal-ksolves (via GitHub)" <gi...@apache.org>.
arvindKandpal-ksolves commented on PR #124:
URL: https://github.com/apache/flink-connector-pulsar/pull/124#issuecomment-4591603364
Hi @featzhang , follow up on this Patch. Please review it whenever have time.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org