Deprecate and replace MessageConverter with reactive IncomingMessageConverter #1772
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
|
@cescoffier As discussed in the Zulip chat. There are a 5 or 6 failing Kafka tests that seem to be a result of the now required |
Codecov Report
@@ Coverage Diff @@
## main #1772 +/- ##
=============================================
- Coverage 73.83% 12.49% -61.34%
+ Complexity 3145 441 -2704
=============================================
Files 275 276 +1
Lines 11228 11238 +10
Branches 1437 1432 -5
=============================================
- Hits 8290 1404 -6886
- Misses 2249 9645 +7396
+ Partials 689 189 -500
|
This covers issues with message conversion failures related to both MANUAL and POST_PROCESSING ack strategies.
### IncomingMessageConverter / MessageConverter * Introduces IncomingMessageConverter which as a reactive interface * * Naming allows/prepares for outgoing message converters as well * Deprecates MessageConverter ### ConverterUtils * `canCovert` is now always called (as the API docs suggest) even after converter caching * Identity converter is no longer used; compatbile payloads and target types use go through no conversion * Handles `MessageConverter.convert` failures * * Wrapped with logging and automatic message rejection Fixes smallrye#1769
The ad-hoc conversion that was happening in `IncomingRabbitMQMessage.getPayload()` has been removed and replaced with a standard set of `IncomingMessageConverter`s that support JSON, text and byte arrays; the same that was attempted in the getPayload. This ensures conversion only happens once (instead of every time getPayload is called) and that errors don’t happen or are handled properly when the do. The removes `content-type-override` option. It was a hack to get the ad-hoc conversion to treat messages as some other type. This should be handled in an `IncomingMessageConverter` with proper error handling (which wasn’t done before.
kdubb
force-pushed
the
fix/message_conversion
branch
from
be40a56
to
a4f059a
Compare
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.
Fixes #1769
IncomingMessageConverter/MessageConverterDeprecate
MesasgeConverterMessageConverter's interface is synchronous which does not allow it to handle failures in any meaningful way. First, if message conversion fails the interface requires it to return something which cannot currently be null; meaning there is no current mechanism to "ignore" a failed message. Second, it is required thatMessageConverters handle callingackornackfor a failure. Unfortunately those methods returnCompletionStages. Converters must ignore the results ofack/nackor perform some sort of blocking wait for them to complete.After talking with @cescoffier we decided changing the interface to use a reactive method would be best.
Introducing IncomingMessageConverter
IncomingMessageConverteris the same interface but returns aUni<Message<?>>. Changing its name serves two purposes. 1) The new name allows/prepares for outgoing message converters (as discussed in issues like #1162). 2) It allows continued use of (now deprecated)MessageConverteruntil such time as it is removed.Adapter AnyMessageConverter
To facilitate using both
IncomingMessageConverterandMessageConvertera new base interfaceAnyMessageConverterwas added. It is marked deprecated along withMessageConverterand when it is decided to removeMessageConverter,AnyMessageConverterwill be removed as well.ConverterUtils Changes
Wrapping Legacy Converters
Calls to the deprecated
MessageConverter.convertare now wrapped with failure handling that logs the failure, rejects (akanacks) and ignores the message. This solves issues with stalling of message handling (e.g. #1769) since failures cannot be handled by these converters.Wrapping is not done for
IncomingMessageConverter, now that the interface provides the full capabilities to handle failures gracefully it is expected that they will. All failures from streams returned fromIncomingMessageConverterare ignored without logging or ack/nack (this is explained in the interface's API docs).canConvertPreviously due to caching of converters, the
canCovertmethod was only ever called one time, when the converter was cached. This has the following problems:convertis called.canConvertis passed an instance ofMessage<?>to make its decision but each call toconvertwill be passed a unique instance ofMessage<?>which may not be anything like the message passed tocanCovert.For these reasons
canConvertis now always called (as the API docs already suggest) even after converter caching. If a converter cannot be used with the current message instance a new converter will be cached. For converters that see the same exact messages all the time there is little to no penalty. Additionally converters that may see a wide range of message types will now work correctly.No
IdentityconverterThe
ConverterUtils.convertmethod checks for the ability to assign the payload type in the message to the type target payload type required by the injection target. There is no need to lookup and use an identity converter since the message payload is already in a form the injection target can use.RabbitMQ Conversion
IncomingRabbitMQMessage's ad-hoc conversion inIncomingRabbitMQMessage.getPayload()has been replaced with a standard set of converts that replicate the previous behavior but with better error handling and more deterministic logic.Additionally this removes the
content-type-overridefor RabbitMQ incoming connectors. It was used as a hack to make conversion happen on unsupported content types. This should be done in anIncomingMessageConverter, with proper error handling, whenever this functionality is needed.