Major Description
Samza uses the Serde interface to serialize and deserialize messages when producing and sending them.
Right now, if a Serde is unable to (de)serialize a message, it has no choice but to throw an exception, or put some non-standard surrogate object in the message envelope's key/value fields. If an exception is throws, the whole container fails.
We should make it possible for Serdes to signal to Samza that they were unable to (de)serialize the message in question, and add configuration to Samza to be able to drop these messages rather than throwing an exception.