[Bug] Acknowledgements lost on bookies and brokers restart resulting in messages not being delivered

[szkoludasebastian]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

Client version: 3.2.2, Server version: 3.2.2
On previous version also notice same behaviour, e.g. 3.1.0, 3.1.2

Minimal reproduce step

Noticed messages loss when we restart all bookies and brokers during processing the data, so data is send to some topic and then our application consumes messages from topic saves msg payload somewhere and acknowledges messages. To be more precise here are steps:

1. Simulate data streaming - send 1000000 msgs to topic A in some time, e.g. 1000000 msgs in 2 minutes.
2. During the time when messages are sent restart all bookies and brokers at once (in our case we have 8 bookies and 6 brokers).
3. In the same time application is consuming messages from topic A, saves message payload somewhere and after successful save acknowledges messages.

What did you expect to see?

No message loss

What did you see instead?

Some messages are lost. So when we send 1000000 messages, in directory where we store messages we see less than 1000000. We can't specify here how much less, because it is a very random situation. Sometimes we have all the messages, but sometimes something is missing.

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[dao-jun]

It looks a little strange, did you handle the send message result?

I mean, what do you do when send message failed?

It looks you don't handle the send message result, if there are some messages failed to send, you should handle the exception and retry again

In general, Pulsar do not retry to send message automatically, it's user's responsibility

[szkoludasebastian]

I think it's not the case, because we are retrying. We handle message result and retry message if it was not sent correctly. During this test which I described we are sending messages till we won't have 1000000 correctly sent.

So we are sure that on input we have 1000000 messages.

[dao-jun]

Did you mean you send 1000000 messages, the results are succeed, but after restart broker/bookie, some of them lost?

[szkoludasebastian]

I mean we send 1000000 messages and during that time when our script was sending messages we restarted all bookies and brokers and then we were able to observe that some of messages lost. It was random situation for us, so we had to perform this test few times to observe that message loss.

[dao-jun]

can I see your script?

[szkoludasebastian]

I simplified our infrastructure a bit just to show our problem. To be more accurate our test script sends data to the application using websocket and our application sends messages to the topic. Our application responds with either ok or error results and in the case of error the script resends the message. In the case of the script and our application we see that as many messages as we expected were sent correctly.

[szkoludasebastian]

We performed many failover tests using that script and we haven't observed any message loss. Only in this case when we restarted bookies and brokers in same time we observed something like that.

[lhotari]

In the case of the script and our application we see that as many messages as we expected were sent correctly.

Which Pulsar client do you use? When you say "sent correctly", how do you define that?

[szkoludasebastian]

We use pulsar client version: 3.2.2 (org.apache.pulsar:pulsar-client-api:3.2.2).

When I say that message was sent correctly I mean that we use something like this:

public CompletableFuture<Response> sendMessage(Request request) {
	    return producer.newMessage()
                            .value(request.getPayload())	    
			    .sendAsync()
			    .thenApply(messageId -> ok(request))
			    .exceptionallyCompose(ex -> createErrorResponse(request, ex));
    }

so we are creating message based on websocket request, then we use sendAsync() and we use thenApply() to return ok response. When any exception appears then we use exceptionallyCompose() and we return error response.

[szkoludasebastian]

Here is our configuration for pulsar client (in comments I put values):

public PulsarClient createPulsarClient(ClientBuilder pulsarClientBuilder, PulsarClientProperties properties) throws PulsarClientException {
        return pulsarClientBuilder
                .serviceUrl(properties.getProxyUrl()) //pulsar://localhost:6650
                .ioThreads(properties.getHandlingBrokersConnectionThreads()) //10
                .listenerThreads(properties.getListenerThreads()) //1
                .connectionsPerBroker(properties.getMaxConnectionsPerBroker()) //10
                .connectionTimeout(parseDurationToSeconds(properties.getConnectionTimeout()), TimeUnit.SECONDS) //PT30S
                .operationTimeout(parseDurationToSeconds(properties.getOperationTimeout()), TimeUnit.SECONDS) //PT30S
                .statsInterval(parseDurationToSeconds(properties.getStatsInterval()), TimeUnit.SECONDS)
                .build();
    }

and also configuration for producer:

public Producer<byte[]> createProducer(PulsarClient pulsarClient, PulsarProducerProperties properties) {

	    ProducerBuilder<byte[]> builder = pulsarClient.newProducer(Schema.BYTES);
	    return builder.topic(properties.getTopicPublish())
                .batchingMaxPublishDelay(properties.getBatchingMaxPublishDelayMicros(), TimeUnit.MICROSECONDS) //100000
                .batcherBuilder(BatcherBuilder.KEY_BASED)
                .hashingScheme(HashingScheme.Murmur3_32Hash)
                .blockIfQueueFull(true)
                .sendTimeout(parseDurationToSeconds(properties.getSendTimeout()), TimeUnit.SECONDS) //PT30S
                .create();
 }

[lhotari]

so we are creating message based on websocket request, then we use sendAsync() and we use thenApply() to return ok response. When any exception appears then we use exceptionallyCompose() and we return error response.

@szkoludasebastian This looks correct.

[lhotari]

Some messages are lost. So when we send 1000000 messages, in directory where we store messages we see less than 1000000. We can't specify here how much less, because it is a very random situation. Sometimes we have all the messages, but sometimes something is missing.

Another point of view is to say that the messages aren't delivered to the consumer in your test scenario. It's about the same as message loss from your application perspective, but there's a subtle difference. Would you be able to check if the messages are stored ok and would be available for delivery on another subscription and consumer that is started after this failure scenario. It's possible that the message loss happens in delivery on the consumer side.

What subscription type are you using? How do you handle acknowledgements?

When the problem happens, please share the internal stats for the topic and the subscription (use pulsar-admin topics stats-internal [topicname] to get it) immediately after the problem has happened. That would reveal useful information about the internal states. Please also provide topic stats ( pulsar-admin topics stats [topicname]), just to be sure that all available information is available for diagnostics.

[szkoludasebastian]

1. Here is consumer configuration:

public Consumer<byte[]> byteConsumer(ConsumerConfigurationProperties consumerConfigurationProperties) throws PulsarClientException {
        return PulsarClient.builder()
                .serviceUrl(consumerConfigurationProperties.getProxyUrl())
                .statsInterval(consumerConfigurationProperties.getStatsInterval(), TimeUnit.SECONDS)
                .build()
                .newConsumer()
                .subscriptionMode(SubscriptionMode.Durable)
                .topic(consumerConfigurationProperties.getTopicSubscribeDecorated())
                .subscriptionName(consumerConfigurationProperties.getSubscriptionName())
                .consumerName(String.format("%s-%s", consumerConfigurationProperties.getConsumerName(), UUID.randomUUID()))
                .subscriptionType(SubscriptionType.Key_Shared)
                .keySharedPolicy(KeySharedPolicy.autoSplitHashRange())
                .subscribe();
    }

So we are using Key_Shared subscription.

2. We are handling acknowledgments this way:

public void acknowledgeMessages(List<DomainMessageId> messages) {
        List<MessageId> messageIds = messages.stream()
                .map(this::toMessageId)
                .toList();

        consumer.acknowledgeAsync(messageIds)
                .exceptionally(throwable -> {
                    log.error("ACKNOWLEDGE ERROR, messages Ids: {}", messageIds, throwable);
                    throw new IllegalStateException(throwable);
                });
    }

public void negativeAcknowledge(List<DomainMessageId> messages) {
        messages.stream()
                .map(this::toMessageId)
                .forEach(consumer::negativeAcknowledge);
    }

So we have these two methods. One for acknowledge messages and second one for negative ack. We consume messages from topic and store them in batches. When it's time to flush batch and it is correctly stored in directory which we want then we are using this method to acknowledge messages by id. When there is some error with storing data in desired directory then we negatively acknowledge messages.

3. Here we have topics partitioned-stats:

{
  "msgRateIn" : 0.0,
  "msgThroughputIn" : 0.0,
  "msgRateOut" : 0.0,
  "msgThroughputOut" : 0.0,
  "bytesInCounter" : 0,
  "msgInCounter" : 0,
  "bytesOutCounter" : 8776881,
  "msgOutCounter" : 13359,
  "averageMsgSize" : 0.0,
  "msgChunkPublished" : false,
  "storageSize" : 262236942,
  "backlogSize" : 0,
  "publishRateLimitedTimes" : 0,
  "earliestMsgPublishTimeInBacklogs" : 0,
  "offloadedStorageSize" : 0,
  "lastOffloadLedgerId" : 0,
  "lastOffloadSuccessTimeStamp" : 0,
  "lastOffloadFailureTimeStamp" : 0,
  "ongoingTxnCount" : 0,
  "abortedTxnCount" : 0,
  "committedTxnCount" : 0,
  "publishers" : [ {
    "msgRateIn" : 0.0,
    "msgThroughputIn" : 0.0,
    "averageMsgSize" : 0.0,
    "chunkedMessageRate" : 0.0,
    "producerId" : 0,
    "supportsPartialProducer" : false
  }, {
    "msgRateIn" : 0.0,
    "msgThroughputIn" : 0.0,
    "averageMsgSize" : 0.0,
    "chunkedMessageRate" : 0.0,
    "producerId" : 0,
    "supportsPartialProducer" : false
  } ],
  "waitingPublishers" : 0,
  "subscriptions" : {
    "microbatcher" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 8776881,
      "msgOutCounter" : 13359,
      "msgRateRedeliver" : 0.0,
      "messageAckRate" : 222.64888172257446,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 0,
      "backlogSize" : 0,
      "earliestMsgPublishTimeInBacklog" : 0,
      "msgBacklogNoDelayed" : 0,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "type" : "Key_Shared",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 0,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "lastMarkDeleteAdvancedTimestamp" : 0,
      "consumers" : [ {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 4947858,
        "msgOutCounter" : 7531,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 125.51649435506889,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 47969,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 3829023,
        "msgOutCounter" : 5828,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 97.13238736750556,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 48422,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 0,
        "msgOutCounter" : 0,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 0.0,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 50000,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 0,
        "msgOutCounter" : 0,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 0.0,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 50000,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      } ],
      "isDurable" : true,
      "isReplicated" : false,
      "allowOutOfOrderDelivery" : false,
      "consumersAfterMarkDeletePosition" : { },
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "delayedMessageIndexSizeInBytes" : 0,
      "subscriptionProperties" : { },
      "filterProcessedMsgCount" : 0,
      "filterAcceptedMsgCount" : 0,
      "filterRejectedMsgCount" : 0,
      "filterRescheduledMsgCount" : 0,
      "durable" : true,
      "replicated" : false
    }
  },
  "replication" : { },
  "nonContiguousDeletedMessagesRanges" : 0,
  "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
  "delayedMessageIndexSizeInBytes" : 0,
  "compaction" : {
    "lastCompactionRemovedEventCount" : 0,
    "lastCompactionSucceedTimestamp" : 0,
    "lastCompactionFailedTimestamp" : 0,
    "lastCompactionDurationTimeInMills" : 0
  },
  "metadata" : {
    "partitions" : 100,
    "deleted" : false
  },
  "partitions" : { }
}

4. Here we have topics partitioned-stats-internal (attached as json file because there is a lot of lines):
   partitioned-stats-internal.json

[szkoludasebastian]

I've added new subscription and consumer but there was no messages:

{
  "msgRateIn" : 0.0,
  "msgThroughputIn" : 0.0,
  "msgRateOut" : 0.0,
  "msgThroughputOut" : 0.0,
  "bytesInCounter" : 0,
  "msgInCounter" : 0,
  "bytesOutCounter" : 8776881,
  "msgOutCounter" : 13359,
  "averageMsgSize" : 0.0,
  "msgChunkPublished" : false,
  "storageSize" : 262236942,
  "backlogSize" : 0,
  "publishRateLimitedTimes" : 0,
  "earliestMsgPublishTimeInBacklogs" : 0,
  "offloadedStorageSize" : 0,
  "lastOffloadLedgerId" : 0,
  "lastOffloadSuccessTimeStamp" : 0,
  "lastOffloadFailureTimeStamp" : 0,
  "ongoingTxnCount" : 0,
  "abortedTxnCount" : 0,
  "committedTxnCount" : 0,
  "publishers" : [ {
    "msgRateIn" : 0.0,
    "msgThroughputIn" : 0.0,
    "averageMsgSize" : 0.0,
    "chunkedMessageRate" : 0.0,
    "producerId" : 0,
    "supportsPartialProducer" : false
  }, {
    "msgRateIn" : 0.0,
    "msgThroughputIn" : 0.0,
    "averageMsgSize" : 0.0,
    "chunkedMessageRate" : 0.0,
    "producerId" : 0,
    "supportsPartialProducer" : false
  } ],
  "waitingPublishers" : 0,
  "subscriptions" : {
    "message-loss-sub" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 0,
      "msgOutCounter" : 0,
      "msgRateRedeliver" : 0.0,
      "messageAckRate" : 0.0,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 0,
      "backlogSize" : 0,
      "earliestMsgPublishTimeInBacklog" : 0,
      "msgBacklogNoDelayed" : 0,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "type" : "Key_Shared",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 0,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "lastMarkDeleteAdvancedTimestamp" : 0,
      "consumers" : [ {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 0,
        "msgOutCounter" : 0,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 0.0,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 50000,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      } ],
      "isDurable" : true,
      "isReplicated" : false,
      "allowOutOfOrderDelivery" : false,
      "consumersAfterMarkDeletePosition" : { },
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "delayedMessageIndexSizeInBytes" : 0,
      "subscriptionProperties" : { },
      "filterProcessedMsgCount" : 0,
      "filterAcceptedMsgCount" : 0,
      "filterRejectedMsgCount" : 0,
      "filterRescheduledMsgCount" : 0,
      "durable" : true,
      "replicated" : false
    },
    "microbatcher" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 8776881,
      "msgOutCounter" : 13359,
      "msgRateRedeliver" : 0.0,
      "messageAckRate" : 0.0,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 0,
      "backlogSize" : 0,
      "earliestMsgPublishTimeInBacklog" : 0,
      "msgBacklogNoDelayed" : 0,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "type" : "Key_Shared",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 0,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "lastMarkDeleteAdvancedTimestamp" : 0,
      "consumers" : [ {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 4947858,
        "msgOutCounter" : 7531,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 0.0,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 47969,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 3829023,
        "msgOutCounter" : 5828,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 0.0,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 48422,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 0,
        "msgOutCounter" : 0,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 0.0,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 50000,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      }, {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 0,
        "msgOutCounter" : 0,
        "msgRateRedeliver" : 0.0,
        "messageAckRate" : 0.0,
        "chunkedMessageRate" : 0.0,
        "availablePermits" : 50000,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 0,
        "blockedConsumerOnUnackedMsgs" : false,
        "readPositionWhenJoining" : "59360:3396",
        "lastAckedTimestamp" : 0,
        "lastConsumedTimestamp" : 0,
        "lastConsumedFlowTimestamp" : 0,
        "lastAckedTime" : "1970-01-01T00:00:00Z",
        "lastConsumedTime" : "1970-01-01T00:00:00Z"
      } ],
      "isDurable" : true,
      "isReplicated" : false,
      "allowOutOfOrderDelivery" : false,
      "consumersAfterMarkDeletePosition" : { },
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "delayedMessageIndexSizeInBytes" : 0,
      "subscriptionProperties" : { },
      "filterProcessedMsgCount" : 0,
      "filterAcceptedMsgCount" : 0,
      "filterRejectedMsgCount" : 0,
      "filterRescheduledMsgCount" : 0,
      "durable" : true,
      "replicated" : false
    }
  },
  "replication" : { },
  "nonContiguousDeletedMessagesRanges" : 0,
  "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
  "delayedMessageIndexSizeInBytes" : 0,
  "compaction" : {
    "lastCompactionRemovedEventCount" : 0,
    "lastCompactionSucceedTimestamp" : 0,
    "lastCompactionFailedTimestamp" : 0,
    "lastCompactionDurationTimeInMills" : 0
  },
  "metadata" : {
    "partitions" : 100,
    "deleted" : false
  },
  "partitions" : { }
}