Catching Exception in Debezium MongoDB Connector [DBZ-6169]

Migrated from [DBZ-6169](https://issues.redhat.com/browse/DBZ-6169)

From [https://debezium.zulipchat.com/#narrow/stream/302529-users/topic/Catch.20exception.20in.20Debezium]

Creating a Jira as kindly suggested by [~ccranfor@redhat.com] – hoping I've chosen Issue Type correctly as a task, as it's not a bug or a feature request. Rather this is more about seeking advice from experienced people here.

I have a scenario where Debezium for MongoDB is working correctly but the source data has an edge case (less than 0.1%) where the nesting is heavily used and a message emitted by MongoDB Change Stream + Debezium connector exceeds Pub/sub 10 MB limit, breaking the whole streaming pipeline.

I'm trying to find a way to catch exception in Debezium for example the following error

io.debezium.server.ConnectorLifecycle - Connector completed: success = 'false', message = 'Stopping connector after error in the application's handler method: java.util.concurrent.ExecutionException: com.google.api.gax.rpc.InvalidArgumentException: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: Request payload size exceeds the limit: 10485760 bytes.', error = 'io.debezium.DebeziumException: java.util.concurrent.ExecutionException: com.google.api.gax.rpc.InvalidArgumentException: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: Request payload size exceeds the limit: 10485760 bytes.

I’m hoping to override the message e.g. just sending the document ID instead of the whole document, so it can be processed separately

Chris shared some insights that:

Chris Cranford: Hi @KS I'm not sure that's necessarily how we'd want to handle that scenario, at least not by default. Now that's not to say we can't add a toggle to implement this behavior but I think there is value in trying to consider an abstraction layer for this.

For example, Debezium's source connectors have this concept of an {{ErrorHandler}} contract where the handler can define how Debezium relays failures back to the runtime framework. Perhaps there is a meaningful reason here to consider a similar pattern on the Debezium Server sink side where we can provide enough context information and you could adjust the behavior, i.e. if the failure is a message payload size too big, you could then resend with just the document id as you described by providing your own custom error handler.

I'd suggest raising a Jira and lets discuss it in more detail there.

The issue names Debezium’s source-side ErrorHandler contract and the Debezium Server sink side, but no implementation file or test. Start by tracing that contract and the sink failure path; done requires an agreed approach for oversized MongoDB messages, including whether the proposed document-ID fallback belongs in a configurable error handler.

Written by the indexing model from the issue text.