code: @SuppressWarnings("unused")
      public void destroy() {
          if (cacheProducers != null) {
              for (Entry<ProducerType, KafkaProducer<String, byte[]>> entry : cacheProducers
                      .entrySet()) {
                  try {
                      KafkaProducer<String, byte[]> producer = entry.getValue();
                      producer.close();
                  } catch (Exception e) {
                      LOGGER.warn(e.getMessage(), e);
                  }
              }
              cacheProducers.clear();
              cacheProducers = null;
          }
  
          try {
              if (metadataConsumer != null) {
                  metadataConsumer.close();
              }
          } catch (Exception e) {
              LOGGER.warn(e.getMessage(), e);
          } finally {
              metadataConsumer = null;
          }
  
          if (cacheConsumers != null) {
              for (Entry<String, KafkaMsgConsumer> entry : cacheConsumers.entrySet()) {
                  try {
                      KafkaMsgConsumer consumer = entry.getValue();
                      consumer.destroy();
                  } catch (Exception e) {
                      LOGGER.warn(e.getMessage(), e);
                  }
              }
              cacheConsumers.clear();
              cacheConsumers = null;
          }
  
          // // to fix the error
          // //
          // "thread Thread[metrics-meter-tick-thread-1...] was interrupted but is
          // still alive..."
          // // since Kafka does not shutdown Metrics registry on close.
          // Metrics.defaultRegistry().shutdown();
  
          if (executorService != null && myOwnExecutorService) {
              try {
                  List<Runnable> tasks = executorService.shutdownNow();
              } catch (Exception e) {
                  LOGGER.warn(e.getMessage(), e);
              } finally {
                  executorService = null;
              }
          }
      }
docs: Destroying method.