apache / apache/pulsar-connectors
[improve][io] JDBC sink: replace synchronized queue with LinkedBlockingDeque for proper back-pressure
- Dominant language
- Java
- Stars
- 26
- Forks
- 25
- Avg merge
- 17h 7m
- Merged PRs (30d)
- 1
Description
## Motivation
This is a follow-up to #9 which introduced bounded queue back-pressure via `record.fail()` when the queue is full. While that prevents OOM, it causes a **nack/redeliver storm** under sustained load: the consumer continuously delivers messages, the sink immediately fails them, they get redelivered, and the cycle repeats — wasting CPU time on rejecting/redelivering messages instead of actual DB writes.
## Changes
### 1. Replace `LinkedList` + `synchronized` with `LinkedBlockingDeque`
Replace the manual `synchronized(incomingList)` blocks with JDK's `LinkedBlockingDeque`, which handles all synchronization internally. This follows the established pattern already used by other connectors in this repo (Aerospike sink uses `LinkedBlockingDeque`, HDFS/Kinesis/DynamoDB sinks use `LinkedBlockingQueue`).
### 2. Blocking back-pressure in `write()`
Instead of `record.fail()` (which triggers immediate nack/redeliver), use `incomingList.offer(record, 1, TimeUnit.SECONDS)` which **blocks the Pulsar IO thread** until space is available. Blocking the IO thread is the correct back-pressure signal — it stops the consumer from fetching more messages for this sink's subscription.
- If space becomes available within 1 second (flush drains the queue), the record is accepted — zero nacks.
- If the timeout expires (flush thread stuck or very slow), the record is failed — at most 1 nack/second vs. thousands/second with the current approach.
### 3. Replace recursive `flush()` with iterative loop
The current `flush()` calls itself recursively when `needAnotherRound` is true. Under sustained load with large queues, this can build up deep stack frames. Replace with a `while` loop.
### 4. Move `isFlushing.set(false)` to `finally` block
Prevents the `isFlushing` flag from getting stuck if an exception occurs before the flag is cleared, which would prevent all future flushes.
### 5. Drain remaining records in `close()`
On shutdown, drain any records still in the queue and fail them, ensuring clean shutdown without message loss.
## Compatibility
- `maxQueueSize` config is fully preserved: `-1` (unbounded), `0` (auto), positive (bounded)
- `LinkedBlockingDeque` with no capacity argument = unbounded (same as legacy `LinkedList`)
- No new configuration parameters required
Contributor guide
No contributing guide indexed for this repository
Research direction
Start at the JDBC sink's write(), flush(), and close() entry points, then compare the queue pattern with the Aerospike sink's LinkedBlockingDeque usage. Done means bounded and unbounded maxQueueSize behavior is preserved, enqueueing applies timed back-pressure, flushing is iterative and exception-safe, and shutdown fails any remaining records.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- database
- Issue type
- Refactor
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 68/100