@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));
            }
        }
    }
@ 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 ) ) ; } } }
订阅topic