如何编写脚本/代码来阻止apache kafka使用者使用消息？

Question

private static Consumer<String, String> createConsumer() {
    final Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);
    props.put(ConsumerConfig.GROUP_ID_CONFIG,"test-consumer-group");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
    // Create the consumer using props.
    consumer = new KafkaConsumer<>(props);
    // Subscribe to the topic.
    consumer.subscribe(Collections.singletonList(TOPIC));
    return consumer;
}

public static void main(String args[]) throws Exception {
    runConsumer();
}

private static void runConsumer(){
    Consumer<String, String> consumer = createConsumer();
    try{
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : consumerRecords) {
                System.out.printf("partition = %d, offset = %d, key = %s, value = %s\n", record.partition(), record.offset(), record.key(), record.value()); 
            }
            consumer.commitAsync();
        }
    }catch(WakeupException wue){
        System.out.println("Wake Up Exception Occured");
        wue.printStackTrace();
    }
    catch(Exception e){
        e.printStackTrace();
    }finally {
        consumer.close();
    }
}`

Answer 1

使用close方法关闭KafKaconsumer只需调用Consumer.close（）；

公共无效关闭（）

关闭使用者，等待最多30秒的默认超时以进行任何需要的清理。如果启用了自动提交，这将在可能的默认超时内提交当前偏移量。有关详细信息，请参见close（long,TimeUnit）。注意，wakeup()不能用于中断关闭。