Kafka离线时Spring无法启动。

可以在Spring Boot应用程序的配置文件中设置Kafka的超时时间，使得当Kafka无法连接时，Spring Boot不会无限地等待，而是在适当的时间后停止尝试连接。以下是示例代码：

在application.properties文件中设置以下属性：

spring.kafka.bootstrap-servers=<kafka_server_host>:<kafka_server_port>
spring.kafka.consumer.group-id=<consumer_group_id>
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.properties.max.poll.interval.ms=300000

其中，max.poll.interval.ms属性设置了在Kafka无法连接的情况下，消费者最大连续轮询时间为5分钟。

此外，您还可以在KafkaListenerContainerFactory中设置以下属性：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.getContainerProperties().setSyncCommits(true);
    factory.getContainerProperties().setIdleEventInterval(60000L);
    factory.getContainerProperties().setPollTimeout(3000L);
    factory.setRecordFilterStrategy(record -> {
        // Filter out messages with null values
        if (record.value() == null) {
            logger.info("Skipping record with null value: {}", record);
            return true;
        }
        return false;
    });
    factory.setRetryTemplate(new RetryTemplate());
    factory.setAutoStartup(false);
    factory.getContainerProperties().setShutdownTimeout(10000L);
    factory.getContainerProperties().setMaxPollInterval(Duration.ofMinutes(5));
    return factory;
}

在上面的代码中，setMaxPollInterval方法的参数也可以设置为Duration.ofSeconds(30)，这样即使Kafka长时间停机，Spring Boot也可以快速响应。