04 | 我应该选择哪种Kafka？

老师，您好，目前我们使用kafka时，使用的监控工具是kafka manager，后面也尝试过 kafka eagle，但是总是感觉这两款工具做的不尽如人意，有的甚至已经很长时间不维护了，老师能不能推荐下好的监控工具呢？

滴滴开源https://github.com/didi/Logi-KafkaManager，是目前市面上最好用的一站式 Kafka 集群指标监控与运维管控平台，欢迎体验交流

最近接触使用到了两种kafka监控软件，一个是 kafka tools ，能够清晰的看到kafka存储结构。一个是 granafa，能看到消费的折线图。感觉棒棒哒~

按照文章说的，我理解现在国内大部分用的都是apache kafka 是这样吧？

请问老师，kafka是否可以调整一个时间窗口内先发生却后到达的错序数据，使其可以按照事件时间戳的正确顺序被消费？

选择哪个Kafka版本，更多取决于项目性质： 1. 如果是非常紧急的项目，优先选择商业版，毕竟花了钱以后，有人support。 2. 如果是研究性质或者时间相对宽松的项目，选择Apache Kafka，可以在和社区不断交流的过程中加深理解，根据项目需求，做一些定制。

一口气看完4篇，有种从入门到放弃得感觉，感觉越来越重，越来越迷茫！ 调整下，调整下，继续坚持下吧，集成得对于菜鸟合适只想用用，如果真得想玩转它控制它还是社区版本。或者有钱整个全套收费得也是一省百省。呵呵

我想问下kafka性能测试工具有哪些？

老师，Client/Broker 跨版本具体会带来什么性能影响呢？Kafka 在1.1版本后做了协议兼容，允许Client/Broker 协议不一致。如果生产上Client/Broker 协议不一致，除了新协议的功能上不兼容，性能上有什么影响呢？比如Client 是1.1而Broker 是2.3，会不会影响比如Zero copy 而导致性能下降？

老师，咨询个问题，最近通过spingboot使用非注解方式配置kafka消费者，每一段时间会出现(Re-)joining group的情况，导致即便少量消息也会堆积直到消费者挂上，出现这种情况的原因大概会有哪些呢， 配置如下： Properties props = new Properties(); props.put("bootstrap.servers", env.getBootserver()); // 每个消费者分配独立的组号 props.put("group.id", env.getGroupId()); // 如果value合法，则自动提交偏移量 props.put("enable.auto.commit", "false"); // 设置多久一次更新被消费消息的偏移量 props.put("auto.commit.interval.ms", "1000"); // 设置会话响应的时间，超过这个时间kafka可以选择放弃消费或者消费下一条消息 props.put("session.timeout.ms", "30000"); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100"); props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 600000); // 自动重置offset props.put("auto.offset.reset", "earliest"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); DefaultKafkaConsumerFactory kafkaConsumerFactory = new DefaultKafkaConsumerFactory(props); ContainerProperties containerProperties = new ContainerProperties(topicName); containerProperties.setMessageListener(this); containerProperties.setPollTimeout(300000); ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(3); executor.initialize(); containerProperties.setConsumerTaskExecutor(executor); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer(kafkaConsumerFactory, containerProperties);