code: @Override
      public void subscribe() {
          waitClientRunning();
          if (!running) {
              return;
          }
  
          if (partition == null) {
              if (kafkaConsumer != null) {
                  kafkaConsumer.subscribe(Collections.singletonList(topic));
              }
              if (kafkaConsumer2 != null) {
                  kafkaConsumer2.subscribe(Collections.singletonList(topic));
              }
          } else {
              TopicPartition topicPartition = new TopicPartition(topic, partition);
              if (kafkaConsumer != null) {
                  kafkaConsumer.assign(Collections.singletonList(topicPartition));
              }
              if (kafkaConsumer2 != null) {
                  kafkaConsumer2.assign(Collections.singletonList(topicPartition));
              }
          }
      }
docs: 订阅topic