code: public void shutdown() {
          for (Topic topic : _topics.values()) {
              if (topic.getConsumerConnector() != null) {
                  topic.getConsumerConnector().shutdown();
              }
              topic.getStreamExecutorService().shutdownNow();
              try {
                  topic.getStreamExecutorService().awaitTermination(60, TimeUnit.SECONDS);
              } catch (InterruptedException e) {
                  _logger.warn("Stream executor service was interrupted while awaiting termination. This should never happen.");
              }
          }
          _logger.debug("Pushing unflushed messages back to Kafka.");
  
          Producer producer = new Producer(_configuration);
  
          for (Map.Entry<String, Topic> entry : _topics.entrySet()) {
              String topicName = entry.getKey();
              Topic topic = entry.getValue();
              List<String> unflushedMessages = new ArrayList<String>();
  
              if (!topic.getMessages().isEmpty()) {
                  topic.getMessages().drainTo(unflushedMessages);
                  producer.enqueue(topicName, unflushedMessages);
              }
              _logger.debug("{} messages for topic {} enqueued on Kafka queue", unflushedMessages.size(), topicName);
          }
          producer.shutdown();
      }
docs: Enqueue un-flushed messages back on to Kafka.