private boolean isFatalException(Exception e) {
    return e instanceof UnsupportedVersionException ||
           e instanceof AuthorizationException ||
           e instanceof ProducerFencedException ||
           e instanceof OutOfOrderSequenceException;
}
 

@Test
public void testKafkaExceptions() {
  //exceptions mapped in KafkaExceptionMapper
  verifyMapperResponse(new BrokerNotAvailableException("some message"), Status.SERVICE_UNAVAILABLE,
      BROKER_NOT_AVAILABLE_ERROR_CODE);

  verifyMapperResponse(new InvalidReplicationFactorException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new SecurityDisabledException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new UnsupportedVersionException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new InvalidPartitionsException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new InvalidRequestException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new UnknownServerException("some message"),Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new UnknownTopicOrPartitionException("some message"), Status.NOT_FOUND,
      KAFKA_UNKNOWN_TOPIC_PARTITION_CODE);
  verifyMapperResponse(new PolicyViolationException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new TopicExistsException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);
  verifyMapperResponse(new InvalidConfigurationException("some message"), Status.BAD_REQUEST,
      KAFKA_BAD_REQUEST_ERROR_CODE);

  //test couple of retriable exceptions
  verifyMapperResponse(new NotCoordinatorException("some message"), Status.INTERNAL_SERVER_ERROR,
      KAFKA_RETRIABLE_ERROR_ERROR_CODE);
  verifyMapperResponse(new NotEnoughReplicasException("some message"), Status.INTERNAL_SERVER_ERROR,
      KAFKA_RETRIABLE_ERROR_ERROR_CODE);

  //test couple of kafka exception
  verifyMapperResponse(new CommitFailedException(), Status.INTERNAL_SERVER_ERROR,
      KAFKA_ERROR_ERROR_CODE);
  verifyMapperResponse(new ConcurrentTransactionsException("some message"), Status.INTERNAL_SERVER_ERROR,
      KAFKA_ERROR_ERROR_CODE);

  //test few general exceptions
  verifyMapperResponse(new NullPointerException("some message"), Status.INTERNAL_SERVER_ERROR,
      Status.INTERNAL_SERVER_ERROR.getStatusCode());
  verifyMapperResponse(new IllegalArgumentException("some message"), Status.INTERNAL_SERVER_ERROR,
      Status.INTERNAL_SERVER_ERROR.getStatusCode());
}