最近遇到kafka消息丢失的偶现问题，排查许久都没找到原因。后面通读代码，才发现消息丢失和seekToEnd有关。
我有一套环境是HA架构，3个节点，每个节点有多个app，每个app启动时会向zk注册，然后利用zk选出主app，zk选出主之后，被选为主的app则有资格作为kafka消息的接收者，根据收到的kafka消息进行相应业务的处理。
偶现问题就是当某个app被zk选为主之后，平台会向其发送“你是主”的消息，但该app却没收到“你是主”的消息。
虽然代码中使用了seekToEnd方法，该方法的意思就是读取最近的一个消息，但问题不是这个方法导致的。而是，注册kafka消费者的时机不对导致的。
以下是导致问题发生的伪代码：
public static void main(String[] args) {try {initLogger();// 连接zkconnect2Zk();// 连接kafkaconnect2Kafka();} catch (Exception e) {e.printStackTrace();log.error("failed to start server", e);}}顺便看下使用seekToEnd的代码是怎么写的：
public class MyAppConsumer implements Runnable {public MyAppConsumer (String gid, List<String> topics) {init(gid, topics);}@Overridepublic void run() {try {consumer = new KafkaConsumer<>(props);consumer.subscribe(this.topics);consumer.seekToEnd(new ArrayList<>());log.info("Start Consumer: {} {}", gid, topics);} catch (Exception e) {}while (isRunning) {try {ConsumerRecords<String, String> records = consumer.poll(100);handleRecords(records);.......} catch (Exception e) {}}}
}由于这个是偶现问题，所以复现不容易。可以通过增加日志打印，在发送“你是主”的消息和app连接kafka成功，变成kafka消费者的地方增加详细的日志打印，以此来确认问题。
这里我们就靠口述问题发生的场景了：当连接kafka的方法(connect2Kafka)在连接zk(connect2Zk)之后，如果zk选主完成，kafka的连接还未成功，则会导致问题发生。因为zk选主完成之后，平台就会向对应的app发送“你是主”的消息，而此时该app还未连接到kafka，还不是kafka的消费者，当连接kafka成功之后，因为使用了seekToEnd方法，因此该app只会读取最新的消息，之前的都丢弃了，那么就永远也收不到“你是主”的消息了。
既然发生问题的原因找到了，那改起来也就很方便了，将连接kafka的方法(connect2Kafka)放在连接zk(connect2Zk)之前就可以了。伪代码如下：
public static void main(String[] args) {try {initLogger();// 连接kafkaconnect2Kafka();// 连接zkconnect2Zk();} catch (Exception e) {e.printStackTrace();log.error("failed to start server", e);}}回头想想，一般我们遇到的偶现问题，就会觉得很头疼，但当哪天心情好的时候，去慢慢梳理一下代码，也许你就会发现，好家伙，自己给自己挖了一个大坑！！！
