KafkaBroker或Zookeeper发生故障时，Consumer应有哪些预期行为？

当 Kafka Broker 或 Zookeeper 发生故障时，Consumer 应该能够检测并处理这些故障。在 Kafka 中，消费者可以使用以下代码来处理该问题：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");

// enable auto commit of offsets
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");

// create consumer instance
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// subscribe to topic
consumer.subscribe(Collections.singletonList("test-topic"));

while (true) {
   try {
       ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
       for (ConsumerRecord<String, String> record : records) {
           System.out.printf("offset = %d, key = %s, value = %s%n", 
                   record.offset(), record.key(), record.value());
       }
   } catch (WakeupException e) {
       // handle wakeup event
   } catch (Exception e) {
       // handle other exceptions
   }
}

在上述代码中，Consumer 实例会尝试从 Kafka Broker 中拉取数据，并在接收数据后对其进行处理。当 Kafka Broker 或 Zookeeper 发生故障时，Consumer 可能会等待超时或抛出异常，应该对这些情况进行特殊处理。为了保持 Consumer 的健壮性，并确保其能够正确处理故障情况，可以使用如下操作：

1. 使用较短的超时时间

在上述代码中，我们使用了 100 毫秒的超时时间来进行数据拉取。如果 Kafka Broker 或 Zookeeper 发生故障，Consumer 可能会等待较长的时间才能检测到故障情况。因此，可以考虑使用较短的超时时间（例如 10 毫秒）来检