记一次pulsar数据丢失排查

news2025/1/13 15:33:20

记一次pulsar数据丢失排查

背景

生产者往pulsar写消息时会有递增的序列号字段，消费端在消费时，会出现序列号断层。当下无法确定是生产端、mq、消费端哪个地方丢失了数据，所以先从生产端进行排查。
生产端的消息发送是通过sendAsync的异步方法Template.sendAsync该方法会返回一个CompletableFuture对象目前是没有对该对象进行检查，所以补上对应的代码逻辑

template.sendAsync(xxx, protocol)
                    .whenCompleteAsync((res, exp) -> {
                        if (exp != null) {
                            log.error("", exp);
                        }
                    });
        }

在运行一段时间后，该异常被触发，可以确定问题在生产者这里了

异常排查

内存限制

运行一段时间后提示Client memory buffer is full，观察源码得知ProducerImpl在发送消息给mq时，会先进行this.canEnqueueRequest(callback, message.getSequenceId(), uncompressedSize)判断消息是否可以发送。
这里判断的是使用的本地内存大小

if (!this.client.getMemoryLimitController().tryReserveMemory((long)payloadSize)) {
                    this.semaphore.ifPresent(Semaphore::release);
                    callback.sendComplete(new PulsarClientException.MemoryBufferIsFullError("Client memory buffer is full", sequenceId));
                    return false;
                }

在ProducerImpl内部会有一个流量控制器，对当前消息的大小进行累加，判断是否超过限制，默认大小为64M。于是乎这边对限制的大小进行调大，在我们创建PulsarClinet时可以指定如下参数设置使用本地内存大小的最大值，这里我将大小调整
memoryLimit(512, SizeUnit.MEGA_BYTES)
重启后不在提示该异常

超时响应

虽然解决了内存问题，新的异常又开始出现
can not send message to the topic persistent://public within given timeout : createdAt 5.207 seconds ago, firstSentAt 5.206 seconds ago, lastSentAt 5.206 seconds ago, retryCount 1
在讲解这个异常怎么检测前，先说说ProducerImpl的发送机制，发送的数据会被加入到ProducerImpl的OpSendMsgQueue对象中
在这里插入图片描述
在pulsar中间件处理完数据后会告诉回发送端，这时ClientCnx netty对象就会回调handleSendReceipt方法将OpSendMsgQueue的数据进行移出，就是告诉发送端这条消息存储成功的一个响应-应答模式

那如果消息一直没有ack，这时发送端就会有个内部定时器进行定时检测，其在ProducerImpl.run方法中，默认执行时间为30s，会判断pendingMessage中的第一条消息是否超过了30s的时间，如果超过了，就会进行failPendingMessages方法调用

                        long diff = this.conf.getSendTimeoutMs() - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - firstMsg.createdAt);
                        if (diff <= 0L) {
                            log.info("[{}] [{}] Message send timed out. Failing {} messages", new Object[]{this.topic, this.producerName, this.pendingMessages.messagesCount()});
                            PulsarClientException te = new PulsarClientException.TimeoutException(String.format("The producer %s can not send message to the topic %s within given timeout", this.producerName, this.topic), firstMsg.sequenceId);
                            this.failPendingMessages(this.cnx(), te);
                            timeToWaitMs = this.conf.getSendTimeoutMs();
                        } else {
                            timeToWaitMs = diff;
                        }