// logs the error and returns CONTINUE
public class LogAndContinueExceptionHandler implements RecordExceptionHandler {...}
 
// logs the error and returns FAIL
public class LogAndFailExceptionHandler implements RecordExceptionHandler {...}
// Then in StreamsConfig.java:
 
.define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
        Type.CLASS,
        LogAndFailExceptionHandler.class.getName(),
        Importance.MEDIUM,
        DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)