Pulsar Sink connectors not getting messages when input rate to pulsar is high

**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](https://pulsar.apache.org/docs/en/io-quickstart/#connect-pulsar-to-cassandra). 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:
1. Setup Pulsar(2.8.1) standalone and Cassandra sink as described in the official docs.
2. Use the Java code given below to insert messages into the Pulsar topic.

```java
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 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 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)
```json
{
"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)
```json
{
"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)
```json
{
"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`!

Start with the Java producer reproduction and the Cassandra sink setup from the Pulsar IO quickstart. Run the topic stats and sink status commands while publishing 100000 messages, then inspect the unackedMessages, availablePermits, numReadFromPulsar, and numWrittenToSink values; done means the sink receives all messages without unexplained retries or accumulating unacked messages.

Written by the indexing model from the issue text.