Remove proxy in DefaultKafkaConsumerFactory #2822
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
The proxy in the `DefaultKafkaConsumerFactory` for `KafkaConsumer` is created only to intercept `close()` call to remove an instance from the `listeners`. There is no need in such a proxy since simple `KafkaConsumer` class extension can handle that scenario. The reason behind this change is to avoid a `Serializable` (`java.lang.reflect.Proxy`) header in the produced message for the `KafkaConsumer` where we still fail to serialize it because other properties of the proxy are not `Serializable` * Introduce `DefaultKafkaConsumerFactory.ExtendedKafkaConsumer` to handle `listeners` interaction * The `createRawConsumer()` might be considered as breaking change since now end-user must extend this `ExtendedKafkaConsumer` to be able to handle `listeners` same way as before * Adjust `DefaultKafkaConsumerFactoryTests.listener()` test for the current code state
sobychacko
approved these changes
Sep 21, 2023
artembilan
added a commit
to artembilan/stream-applications
that referenced
this pull request
Sep 22, 2023
Fixes spring-cloud#500 When `listeners` are provided for `DefaultKafkaConsumerFactory`, the target `KafkaConsumer` instance is proxied. The `java.lang.reflect.Proxy` is `Serializable`, but the value it is wrapping is not. When the `MessageHeaders` is serialized (e.g. into persistent `MessageStore`), it checks for `Serializable` type only on top-level object of the header. Therefore, the `Proxy` is passing condition, but eventually we fail with `NotSerializableException`, since the proxied object is not like that * Remove `kafka_consumer` from a message before it reaches an aggregator with its logic to serialize message into the store This is a workaround until Spring for Apache Kafka is released with the fix: spring-projects/spring-kafka#2822
onobc
pushed a commit
to spring-cloud/stream-applications
that referenced
this pull request
Sep 22, 2023
Fixes #500 When `listeners` are provided for `DefaultKafkaConsumerFactory`, the target `KafkaConsumer` instance is proxied. The `java.lang.reflect.Proxy` is `Serializable`, but the value it is wrapping is not. When the `MessageHeaders` is serialized (e.g. into persistent `MessageStore`), it checks for `Serializable` type only on top-level object of the header. Therefore, the `Proxy` is passing condition, but eventually we fail with `NotSerializableException`, since the proxied object is not like that * Remove `kafka_consumer` from a message before it reaches an aggregator with its logic to serialize message into the store This is a workaround until Spring for Apache Kafka is released with the fix: spring-projects/spring-kafka#2822
garyrussell
suggested changes
Sep 25, 2023
| @@ -445,68 +440,68 @@ private void checkInaccessible(Properties properties, Map<String, Object> modifi | |||
| } | |||
| } | |||
|
|
|||
| @SuppressWarnings("resource") | |||
| protected Consumer<K, V> createKafkaConsumer(Map<String, Object> configProps) { | |||
| checkBootstrap(configProps); | |||
| Consumer<K, V> kafkaConsumer = createRawConsumer(configProps); | |||
There was a problem hiding this comment.
Perhaps we could throw an exception here, if listeners are present and the created consumer is not an instance of ExtendedKafkaConsumer?
There was a problem hiding this comment.
On the other hand, Java doesn't support multiple inheritance (the user might already be extending some other class), so a hard failure would not be so good; perhaps log a warning that listeners must be invoked when closing a custom consumer.
Or, simply revert the code and have ExtendedKafkaConsumer wrap the proxy instead so that the consumer is not Serializable.
But, of course, that would mean we'd have to delegate all methods - ugh!
There was a problem hiding this comment.
I wish we have never had this protected Consumer<K, V> createKafkaConsumer() on our default CF: for any custom client you always can implement your own CF.
So, I'm just going to log a WARN that listeners are ignored because not an ExtendedKafkaConsumer...
There was a problem hiding this comment.
I agree; makes sense; maybe also deprecate that method?
There was a problem hiding this comment.
I would let it live for now until we really fail for some peculiar use-case to learn from.
If we deprecate and remove it (I guess there was a reason behind that method), we may never learn the mentioned use-case 😉
garyrussell
approved these changes
Oct 4, 2023
garyrussell
reviewed
Oct 4, 2023
spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
Outdated
Show resolved
Hide resolved
garyrussell
changed the title
artembilan
added a commit
to spring-cloud/spring-functions-catalog
that referenced
this pull request
Nov 22, 2023
Fixes spring-cloud/stream-applications#500 When `listeners` are provided for `DefaultKafkaConsumerFactory`, the target `KafkaConsumer` instance is proxied. The `java.lang.reflect.Proxy` is `Serializable`, but the value it is wrapping is not. When the `MessageHeaders` is serialized (e.g. into persistent `MessageStore`), it checks for `Serializable` type only on top-level object of the header. Therefore, the `Proxy` is passing condition, but eventually we fail with `NotSerializableException`, since the proxied object is not like that * Remove `kafka_consumer` from a message before it reaches an aggregator with its logic to serialize message into the store This is a workaround until Spring for Apache Kafka is released with the fix: spring-projects/spring-kafka#2822
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
The proxy in the
DefaultKafkaConsumerFactoryforKafkaConsumeris created only to interceptclose()call to remove an instance from thelisteners.There is no need in such a proxy since simple
KafkaConsumerclass extension can handle that scenario.The reason behind this change is to avoid a
Serializable(java.lang.reflect.Proxy) header in the produced message for theKafkaConsumerwhere we still fail to serialize it because other properties of the proxy are notSerializableDefaultKafkaConsumerFactory.ExtendedKafkaConsumerto handlelistenersinteractioncreateRawConsumer()might be considered as breaking change since now end-user must extend thisExtendedKafkaConsumerto be able to handlelistenerssame way as beforeDefaultKafkaConsumerFactoryTests.listener()test for the current code state