public static void handleDeserializationFailure(final DeserializationExceptionHandler deserializationExceptionHandler,
                                                    final ProcessorContext<?, ?> processorContext,
                                                    final Exception deserializationException,
                                                    final ConsumerRecord<byte[], byte[]> rawRecord,
                                                    final Logger log,
                                                    final Sensor droppedRecordsSensor) {
        final DeserializationExceptionHandler.DeserializationHandlerResponse response;
        try {
            response = deserializationExceptionHandler.handle(
                (InternalProcessorContext<?, ?>) processorContext,
                rawRecord,
                deserializationException);
        } catch (final Exception fatalUserException) {
            log.error(
                "Deserialization error callback failed after deserialization error for record {}",
                rawRecord,
                deserializationException);
            throw new StreamsException("Fatal user code error in deserialization error callback", fatalUserException);
        }
        if (response == DeserializationExceptionHandler.DeserializationHandlerResponse.FAIL) {
            throw new StreamsException("Deserialization exception handler is set to fail upon" +
                " a deserialization error. If you would rather have the streaming pipeline" +
                " continue after a deserialization error, please set the " +
                DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG + " appropriately.",
                deserializationException);
        } else {
            log.warn(
                "Skipping record due to deserialization error. topic=[{}] partition=[{}] offset=[{}]",
                rawRecord.topic(),
                rawRecord.partition(),
                rawRecord.offset(),
                deserializationException
            );
            droppedRecordsSensor.record();
        }
    }
@throws StreamsException if a deserialization error occurs and the deserialization callback returns
                         {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL FAIL}
                         or throws an exception itself
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java