apache / apache/pulsar-connectors

[improve][io] JDBC sink: replace synchronized queue with LinkedBlockingDeque for proper back-pressure

Open
#16 0 comments 0 reactions 0 assignees View on GitHub
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.