[Enhancement] Add Backpressure Handling in Kafka Connector Adaptor to Prevent OOM Under Sink Stall Conditions

[sandeep-mst]

Search before reporting

• 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.

User environment

master

Issue Description

The KafkaConnectAdapter currently lacks an effective backpressure mechanism to regulate message ingestion and buffering. As a result, it continues consuming records and appending them to the pendingFlushQueue even when downstream processing is saturated or stalled.

At present, batchSize is only used as a flush threshold and does not impose any limit on writing records. This leads to accumulation of records in memory, which can result in OutOfMemoryError (OOM) under sustained load or when the sink is slow/unavailable.

Error messages

Reproducing the issue

Run any sink using KafkaConnectAdapter (e.g., BigQuery sink) and simulate a slow or stalled downstream. Observe unbounded growth of pendingFlushQueue leading to OOM.

Additional information

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!