[improve][io] JDBC sink: replace synchronized queue with LinkedBlockingDeque for proper back-pressure - #17
Conversation
…ngDeque for proper back-pressure Replaces manual `synchronized(incomingList)` blocks with JDK's `LinkedBlockingDeque` for proper blocking back-pressure instead of the nack/redeliver storm caused by `record.fail()`. Key changes: - Use `offer(record, 1s timeout)` in write() to block the Pulsar IO thread when queue is full, stopping message consumption naturally - Use `drainTo()` in flush() for non-blocking atomic batch drain - Replace recursive flush() with iterative while loop - Move isFlushing.set(false) to finally block - Drain remaining records in close() for clean shutdown Follows the established pattern used by Aerospike, HDFS, Kinesis, and DynamoDB connectors in this repository. Fixes apache#16
There was a problem hiding this comment.
Pull request overview
This PR updates the JDBC sink’s internal buffering/flush mechanics to provide proper back-pressure under sustained load by replacing a manually synchronized LinkedList queue with a bounded LinkedBlockingDeque, and by making flush() iterative and exception-safe.
Changes:
- Replace
LinkedList+synchronizedqueue management withLinkedBlockingDeque(bounded whenmaxQueueSize > 0), using timed blockingoffer()for back-pressure. - Refactor
flush()from recursive to iterative draining (drainTo) and ensureisFlushingis always cleared viafinally. - Drain and fail any remaining queued records during
close()for cleaner shutdown behavior.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| state.set(State.CLOSED); | ||
| // Fail any records still in the queue | ||
| List<Record<T>> remaining = new ArrayList<>(); | ||
| incomingList.drainTo(remaining); | ||
| remaining.forEach(Record::fail); |
| if (!accepted) { | ||
| if (state.get() != State.OPEN) { | ||
| log.warn("Sink became {} while waiting for queue space, failing record", state.get()); | ||
| } else { | ||
| log.warn("Queue still full after timeout (capacity: {}), failing record", maxQueueSize); | ||
| } | ||
| record.fail(); | ||
| return; | ||
| } | ||
|
|
||
| int number = incomingList.size(); |
| state.set(State.CLOSED); | ||
| // Fail any records still in the queue | ||
| List<Record<T>> remaining = new ArrayList<>(); | ||
| incomingList.drainTo(remaining); | ||
| remaining.forEach(Record::fail); |
| if (!accepted) { | ||
| if (state.get() != State.OPEN) { | ||
| log.warn("Sink became {} while waiting for queue space, failing record", state.get()); | ||
| } else { | ||
| log.warn("Queue still full after timeout (capacity: {}), failing record", maxQueueSize); | ||
| } | ||
| record.fail(); | ||
| return; | ||
| } | ||
|
|
||
| int number = incomingList.size(); | ||
| if (batchSize > 0 && number >= batchSize) { |
david-streamlio
left a comment
david-streamlio
left a comment
There was a problem hiding this comment.
The back-pressure design looks right (and matches the Aerospike/Kinesis/DynamoDB connectors), but CI caught a real defect in the new close():
java.lang.NullPointerException: Cannot invoke "LinkedBlockingDeque.drainTo(java.util.Collection)" because "this.incomingList" is null
at org.apache.pulsar.io.jdbc.JdbcAbstractSink.close(JdbcAbstractSink.java:182)
at org.apache.pulsar.io.jdbc.PostgresJdbcArrayIntegrationTest.tearDown(PostgresJdbcArrayIntegrationTest.java:70)
incomingList is only assigned in open(), so close() NPEs whenever the sink is closed without a successful open (which Pulsar IO can do on startup failure, and tests do in tearDown). Please guard the drain:
if (incomingList != null) {
List<Record<T>> remaining = new ArrayList<>();
incomingList.drainTo(remaining);
remaining.forEach(Record::fail);
}Full run: https://github.com/apache/pulsar-connectors/actions/runs/29039246302 (the earlier ProcessedFileThreadTest failure on the first attempt was an unrelated file-connector flake — a fix for that is in progress separately).
|
Heads-up, @harangozop: this PR is now The review finding still stands independently of the conflict. CI caught a real defect in the new
if (incomingList != null) {
List<Record<T>> remaining = new ArrayList<>();
incomingList.drainTo(remaining);
remaining.forEach(Record::fail);
}One thing worth knowing when you rebase: #39 replaced the recursive Happy to help with the rebase if useful. |
Motivation
Follow-up to #9. The
record.fail()back-pressure introduced in #9 causes a nack/redeliver storm under sustained load — the consumer delivers messages, the sink immediately fails them, they get redelivered, repeating endlessly. This wastes CP time.Modifications
Replace
LinkedList+synchronizedwithLinkedBlockingDeque:write():offer(record, 1, TimeUnit.SECONDS)blocks the Pulsar IO thread when queue is full — this IS the back-pressure (stops consumption). At most 1 nack/second on timeout vs. thousands/second before.flush():drainTo(swapList, batchSize)— non-blocking atomic drain, no synchronized blocks. Automatically wakes blockedoffer()calls by making space.flush()uses awhileloop instead of recursive self-calls.isFlushinginfinally: prevents stuck flag after exceptions.close()drains queue: fails remaining records for clean shutdown.Follows the pattern already used by Aerospike (
LinkedBlockingDeque), HDFS, Kinesis, and DynamoDB (LinkedBlockingQueue) connectors in this repo.Verifying this change
testBoundedQueueBackPressuremay need timeout adjustment (6th write now blocks 1s instead of failing instantly)Fixes #16