@Override
  public DeserializationHandlerResponse handle(
      final ProcessorContext context,
      final ConsumerRecord<byte[], byte[]> record,
      final Exception exception
  ) {
    log.debug(
        String.format("Exception caught during Deserialization, "
            + "taskId: %s, topic: %s, partition: %d, offset: %d",
        context.taskId(), record.topic(), record.partition(), record.offset()),
        exception
    );

    streamsErrorCollector.recordError(record.topic());

    if (isCausedByAuthorizationError(exception)) {
      log.info(
          String.format(
              "Authorization error when attempting to access the schema during deserialization. "
              + "taskId: %s, topic: %s, partition: %d, offset: %d",
          context.taskId(), record.topic(), record.partition(), record.offset()));
      return DeserializationHandlerResponse.FAIL;
    }

    return DeserializationHandlerResponse.CONTINUE;
  }
@Test
  public void shouldReturnFailForAuthorizationExceptions() {
    assertThat(exceptionHandler.handle(context, record,
            new Exception("", new RestClientException("", 403, 40301))),
        equalTo(DeserializationHandlerResponse.FAIL));
  }