ISSUE-12747: Pulsar Sink connectors not getting messages when input rate to pulsar is high

Original Issue: apache/pulsar#12747

Describe the bug The Pulsar Sink connector doesn't receive all the messages if the input rate to the pulsar topic is high. I tried using the inbuilt Cassandra connector and followed the instructions to try it out here. I push 100000 messages to pulsar via a java code but I don't see all the messages there in the Sink. Topic receives all the messages but the Sink doesn't. If I send these many messages in small chunks, it reaches the sink.

To Reproduce Steps to reproduce the behavior:

Setup Pulsar(2.8.1) standalone and Cassandra sink as described in the official docs.
Use the Java code given below to insert messages into the Pulsar topic.

import org.apache.pulsar.client.api.BatcherBuilder;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class PushToCassandra {
    public static void main(String[] args) throws PulsarClientException, ExecutionException, InterruptedException {
        /**
         * for i in {0..9}; do bin/pulsar-client produce -m "key-$i" -n 1 test_cassandra; done
         */
        ClientConfigurationData clientConfigurationData = new ClientConfigurationData();
        clientConfigurationData.setServiceUrl("pulsar://localhost:6650");
        PulsarClient pulsarClient = new PulsarClientImpl(clientConfigurationData);
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .topic("test_cassandra")
                .blockIfQueueFull(true)
                .enableBatching(true)
                .batchingMaxMessages(50_000)
                .maxPendingMessages(50_000)
                .batchingMaxBytes(5242880)
                .batchingMaxPublishDelay(10, TimeUnit.SECONDS)
                .batcherBuilder(BatcherBuilder.DEFAULT)
                .create();
        try (producer) {
            Future<MessageId> future = null;
            for (int i = 0; i < 100000; i++) {
                future = producer.newMessage().value("key-" + i).sendAsync();
            }
            producer.flush();
            future.get();
        }
        producer.close();
        pulsarClient.close();
    }
}

Expected behavior A sink should receive all 100000 messages.

Metrics

Topic stats (bin/pulsar-admin topics stats public/default/test_cassandra)

{
"msgRateIn" : 0.0,
"msgThroughputIn" : 0.0,
"msgRateOut" : 0.0,
"msgThroughputOut" : 0.0,
"bytesInCounter" : 1872490,
"msgInCounter" : 100000,
"bytesOutCounter" : 20293526,
"msgOutCounter" : 1100000,
"averageMsgSize" : 0.0,
"msgChunkPublished" : false,
"storageSize" : 1872490,
"backlogSize" : 1872490,
"offloadedStorageSize" : 0,
"lastOffloadLedgerId" : 0,
"lastOffloadSuccessTimeStamp" : 0,
"lastOffloadFailureTimeStamp" : 0,
"publishers" : [ ],
"waitingPublishers" : 0,
"subscriptions" : {
"public/default/cassandra-test-sink" : {
  "msgRateOut" : 0.0,
  "msgThroughputOut" : 0.0,
  "bytesOutCounter" : 20293526,
  "msgOutCounter" : 1100000,
  "msgRateRedeliver" : 0.0,
  "chunkedMessageRate" : 0,
  "msgBacklog" : 2,
  "backlogSize" : 0,
  "msgBacklogNoDelayed" : 2,
  "blockedSubscriptionOnUnackedMsgs" : false,
  "msgDelayed" : 0,
  "unackedMessages" : 50000,
  "type" : "Shared",
  "msgRateExpired" : 0.0,
  "totalMsgExpired" : 0,
  "lastExpireTimestamp" : 0,
  "lastConsumedFlowTimestamp" : 1636623672908,
  "lastConsumedTimestamp" : 1636623672913,
  "lastAckedTimestamp" : 0,
  "lastMarkDeleteAdvancedTimestamp" : 0,
  "consumers" : [ {
    "msgRateOut" : 0.0,
    "msgThroughputOut" : 0.0,
    "bytesOutCounter" : 20293526,
    "msgOutCounter" : 1100000,
    "msgRateRedeliver" : 0.0,
    "chunkedMessageRate" : 0.0,
    "consumerName" : "c958f",
    "availablePermits" : -49000,
    "unackedMessages" : 50000,
    "avgMessagesPerEntry" : 45175,
    "blockedConsumerOnUnackedMsgs" : true,
    "lastAckedTimestamp" : 0,
    "lastConsumedTimestamp" : 1636623672913,
    "metadata" : {
      "instance_id" : "0",
      "application" : "pulsar-sink",
      "instance_hostname" : "aerospike-ThinkPad-P53",
      "id" : "public/default/cassandra-test-sink"
    },
    "connectedSince" : "2021-11-11T14:41:38.541+05:30",
    "address" : "/127.0.0.1:35044",
    "clientVersion" : "2.8.1"
  } ],
  "isDurable" : true,
  "isReplicated" : false,
  "consumersAfterMarkDeletePosition" : { },
  "nonContiguousDeletedMessagesRanges" : 0,
  "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
  "durable" : true,
  "replicated" : false
}
},
"replication" : { },
"deduplicationStatus" : "Disabled",
"nonContiguousDeletedMessagesRanges" : 0,
"nonContiguousDeletedMessagesRangesSerializedSize" : 0,
"compaction" : {
"lastCompactionRemovedEventCount" : 0,
"lastCompactionSucceedTimestamp" : 0,
"lastCompactionFailedTimestamp" : 0,
"lastCompactionDurationTimeInMills" : 0
}
}

Sink stats (bin/pulsar-admin sinks status --tenant public --namespace default --name cassandra-test-sink)

{
"numInstances" : 1,
"numRunning" : 1,
"instances" : [ {
"instanceId" : 0,
"status" : {
  "running" : true,
  "error" : "",
  "numRestarts" : 0,
  "numReadFromPulsar" : 50000,
  "numSystemExceptions" : 0,
  "latestSystemExceptions" : [ ],
  "numSinkExceptions" : 0,
  "latestSinkExceptions" : [ ],
  "numWrittenToSink" : 50000,
  "lastReceivedTime" : 1636621993191,
  "workerId" : "c-standalone-fw-localhost-8080"
}
} ]
}

EDIT

Sink stats after 10 minutes (no change in the intermittent stat check queries)

{
"numInstances" : 1,
"numRunning" : 1,
"instances" : [ {
"instanceId" : 0,
"status" : {
  "running" : true,
  "error" : "",
  "numRestarts" : 0,
  "numReadFromPulsar" : 950000,
  "numSystemExceptions" : 0,
  "latestSystemExceptions" : [ ],
  "numSinkExceptions" : 0,
  "latestSinkExceptions" : [ ],
  "numWrittenToSink" : 950000,
  "lastReceivedTime" : 1636623433172,
  "workerId" : "c-standalone-fw-localhost-8080"
}
} ]
}

availablePermits is also a negative number in the topic stats. Isn't that wrong too? If I run the sink stats check again, the numReadFromPulsar and numWrittenToSink keep on increasing. Now I see 1400000 for them. Pulsar seems to be retrying 50k unackedMessages. Not sure why they are unackedMessages!

streamnative / pulsar-archived

ISSUE-12747: Pulsar Sink connectors not getting messages when input rate to pulsar is high #3255