Adds support for failure message during serialization and handling #26
Conversation
| return; | ||
| } | ||
|
|
||
| $consumer = $this->getConsumer(); |
There was a problem hiding this comment.
can we just call ack() here?
There was a problem hiding this comment.
@KonstantinCodes this depends on the question below, if there is always an envelope with correct serialized class it can use the ack() if you think its a good idea to catch the serialization exception and ack the message + continue then its not possible.
| 'headers' => $message->headers, | ||
| ]); | ||
| } catch (MessageDecodingFailedException $exception) { | ||
| $this->rejectKafkaMessage($message); |
There was a problem hiding this comment.
The default behavior would be that message offsets get committed if the serializer throws an exception.
This would probably be an unexpected side effect.
There was a problem hiding this comment.
@KonstantinCodes this is also done in other transports like RabbitMQ, all enqueue integrations.
The question is now: if serialization fails what should it do? Stop the consumer everytime and try serialization the message everytime (untill the code is fixed to accept this message)?
Or do what it does now: try to deserialize and if it fails throw this exception in the log and ack the message (so no hammering will occur).
There was a problem hiding this comment.
The serializer throws an exception, either because of a programming error or because of an unexpected value. Maybe a new schema, maybe it's an avro message and the schema registry is temporarily down.
In that case, no messages will get processed but they will all be acked.
Once the schema registry is back online, messages will be processed again. We may just have lost a few hours of data. And the programmers have to manually reset the group.id to the last offset before the schema registry went offline, if they want to process the "lost" messages. Additionally, the messages that already got processed after the registry came back online, will be processed again after our group.id reset.
It is a very messy problem to fix.
On the other hand - if message processing waits until the schema registry has come online again, no manual intervention by the programmers is necessary. The system "healed" itself.
It seems to me that the program you have in mind, operates in a "at most once processing" scenario, where it is allowed to more or less randomly skip messages. You're probably dealing with tons of messages and also want to be careful, not to overload the brokers with too many requests.
What's the cause of hammering? In your deployment, how does the symfony consumer process get managed? Does it get restarted immediately after a failure? This could actually create chaos. If you have 3 consumers constantly restarting, they will really hammer Kafka with rebalance operations. You almost DOS your Kafka in that scenario. Maybe an incremental backoff would be the way to go?
Or, if you really operate in "atmost once", why throw an exception in your serializer at all? Just return an empty message. For example, you build a try-catch around your decode logic and in the case of an exception, return a new NullMessage(). Then configure the symfony messenger bus to default_middleware: allow_no_handlers.
In your serializer you could even send messages that failed to decode to a different bus for later re-processing.
There was a problem hiding this comment.
Thanks for your reply!
Im still doubting what to do about the failing serialization, it could have multiple issues:
- Registered schema does not completely match with code so errors could occur durring deserialization (logic thing that this should be fixed)
- There is no registered schema so 'anything' can be send on producer side and consume can fail
- Avro message and schema registry is down
By default the Symfony messenger setup is done with supervisord, so when a consumer stops it just restarts so when a message failed it can just consume the next if you setup the retry strategy correctly.
On all projects at the moment we do now;
- When serialization fails -> stop consumer and retry (untill now this doesn't occured yet)
- When consuming fails (during handling) -> move to failed db transport (able to retry)
But I see a problem arrises when serialization will fail for a single message and hold all the other thousands of messages in Kafka. Of course there is also a problem if all messages fails during serialization (because of down time schema registry or an incorrect code update on producer/consuming side).
There was a problem hiding this comment.
https://github.com/symfony/symfony-docs/pull/13597/files
This PR looks rather interesting. There is a bash script that sleeps for 30 seconds to prevent hammering.
Not sure if this could help but maybe worth a try?
There was a problem hiding this comment.
Yeah sorry, I didn't see your reply yet so I just edited my message #26 (comment)
Basically I'd try to catch the exception in the serializer and move the message the the failed bus.
There was a problem hiding this comment.
@KonstantinCodes indeed true. Good thing to check out consuming and supervisord 👍
Idea to change this PR to not ack when serialization fails (logic should then be in serialization class if it should do something and if it fails it will be retried?) and ack the message if it during handling fails and after that the Kafka can be recreated through retry strategy (eg move to a queue where you can delay a message for retrying like SQS/db/Redis/RabbitMQ) or to the bottom of the Kafka topic (if this is enabled via config)
And set default ack when fail to false
| $consumer->commit($message); | ||
| } | ||
|
|
||
| $this->logger->warning(sprintf('Message %s %s %s rejected successful.', $message->topic_name, $message->partition, $message->offset)); |
There was a problem hiding this comment.
The log line here suggests that the offset hasn't been committed, but it just was committed.
| @@ -70,7 +70,8 @@ public function createTransport(string $dsn, array $options, SerializerInterface | |||
| $options['topic']['name'], | |||
| $options['flushTimeout'] ?? 10000, | |||
| $options['receiveTimeout'] ?? 10000, | |||
| $options['commitAsync'] ?? false | |||
| $options['commitAsync'] ?? false, | |||
| $options['ackFailedMessages'] ?? true | |||
There was a problem hiding this comment.
Can we actually set the default to false here? The reject() call is intended for messages that have failed to process after all.
The Exceptions that cause Messenger to call reject() should be caused by the applications message handlers.
Could the same effect be achieved by the application implementing a "CatchAllExceptionsMiddleware"?
|
Hi @rdotter! Great to see your PR :) I know we talked about this feature a while ago, and your case definitely makes sense. Still, I'm a bit unsure about basically making It really strikes me like the same could be achieved with a |
This solves: #22
When a message is received and there is an MessageDecodingFailedException thrown during serialization or an exception during handling a WorkerMessageFailedEvent will be dispatched.
@KonstantinCodes