code: @Override
      public void unsubscribe() {
          waitClientRunning();
          if (!running) {
              return;
          }
  
          if (kafkaConsumer != null) {
              kafkaConsumer.unsubscribe();
          }
          if (kafkaConsumer2 != null) {
              kafkaConsumer2.unsubscribe();
          }
      }
docs: 取消订阅