@Override
  public synchronized void cleanupAll() throws KafkaResourceManagerException {
    LOG.info("Attempting to cleanup Kafka manager.");

    boolean producedError = false;

    // First, delete kafka topics if it was not given as a static argument
    try {
      if (!usingStaticTopic) {
        kafkaClient.deleteTopics(topicNames).all().get();
      }
    } catch (Exception e) {
      LOG.error("Failed to delete kafka topic.", e);
      producedError = true;
    }

    // Throw Exception at the end if there were any errors
    if (producedError) {
      throw new KafkaResourceManagerException(
          "Failed to delete resources. Check above for errors.");
    }

    super.cleanupAll();

    LOG.info("Kafka manager successfully cleaned up.");
  }
@Test
  public void testCleanupAllShouldThrowErrorWhenKafkaClientFailsToDeleteTopic()
      throws ExecutionException, InterruptedException {
    when(kafkaClient.deleteTopics(anyCollection()).all().get())
        .thenThrow(new ExecutionException(new RuntimeException("delete topic future fails")));

    assertThrows(KafkaResourceManagerException.class, () -> testManager.cleanupAll());
  }