[fix][io] JDBC sink: prevent OOM from unbounded queue on connection failure

[harangozop]

Motivation

Fixes apache/pulsar#25030

The JDBC sink's internal queue (incomingList) is an unbounded LinkedList. When the database connection drops:

1. executeBatch() hangs until TCP socket timeout (can be minutes)
2. isFlushing stays true, blocking all queue draining
3. write() keeps accepting records without any limit
4. incomingList grows until OutOfMemoryError: Java heap space

This is a production issue observed with both PostgreSQL and MariaDB JDBC sinks under Pulsar FunctionMesh.

Modifications

JdbcAbstractSink.java:

• Bounded queue: write() rejects records when queue exceeds maxQueueSize, applying Pulsar-level back-pressure via record.fail() (negative ack → consumer redelivers later)
• State check in write(): records are failed immediately when state != OPEN (after fatal() or close())
• Connection validation & reconnection: ensureConnection() validates the JDBC connection via Connection.isValid() before each flush and reconnects automatically, allowing recovery from transient DB outages without pod restart
• Scheduled flush cancellation: fatal() and close() cancel the periodic flush ScheduledFuture to stop repeated failures on a broken connection

JdbcSinkConfig.java:

• Added maxQueueSize config (default: 0 = auto, which resolves to batchSize * 10). Users can override for their workload.

Verifying this change

• Added testWriteRejectsRecordsAfterFatal — verifies write() fails records when state is FAILED
• Added testBoundedQueueBackPressure — verifies records are failed when internal queue is full
• All existing SQLite sink tests pass (including batch variant)

Does this pull request potentially affect one of the following parts of the Pulsar project?

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment (configuration default changes are backward compatible — maxQueueSize=0 preserves legacy unbounded behavior)

Documentation

• doc-not-needed — Bug fix with backward-compatible config addition

[Copilot]

Pull request overview

This PR addresses a production OOM scenario in the Pulsar JDBC sink when the database connection is slow/broken by adding queue back-pressure, improving sink state handling, and adding connection validation/reconnect behavior.

Changes:

• Add maxQueueSize to bound the internal record queue (with auto-sizing behavior) and reject records when full.
• Fail records immediately when the sink is not OPEN, and cancel periodic flush tasks on fatal() / close().
• Add ensureConnection() to validate the JDBC connection before flushing and reconnect when invalid; add SQLite tests for the new behaviors.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.

File	Description
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java	Adds bounded-queue back-pressure, state checks in write(), scheduled flush cancellation, and connection validation/reconnect in flush().
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcSinkConfig.java	Introduces maxQueueSize configuration and its FieldDoc.
jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java	Adds tests covering failing writes after fatal, bounded-queue back-pressure, and reconnection on invalid connection.

Comments suppressed due to low confidence (1)

jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java:233

• There is a race between the initial state check and enqueuing/scheduling: close()/fatal() can flip state to CLOSED/FAILED and shut down the flushExecutor after the check at line 208, causing write() to enqueue a record and then throw RejectedExecutionException when scheduling flush (or leave a queued record that will never be acked/failed because flush() returns when CLOSED). Re-check state (and executor availability) under the same synchronization used for enqueuing, and handle schedule rejection by failing the record (or flushing synchronously).

    public void write(Record<T> record) throws Exception {
        if (state.get() != State.OPEN) {
            log.warn("Sink is not in OPEN state (current: {}), failing record", state.get());
            record.fail();
            return;
        }
        int number;
        synchronized (incomingList) {
            if (maxQueueSize > 0 && incomingList.size() >= maxQueueSize) {
                if (!queueFullLogged) {
                    log.warn("Internal queue is full ({} >= {}), failing records to apply back-pressure",
                            incomingList.size(), maxQueueSize);
                    queueFullLogged = true;
                }
                record.fail();
                return;
            }
            incomingList.add(record);
            number = incomingList.size();
        }
        if (batchSize > 0 && number >= batchSize) {
            if (log.isDebugEnabled()) {
                log.debug("flushing by batches, hit batch size {}", batchSize);
            }
            flushExecutor.schedule(this::flush, 0, TimeUnit.MILLISECONDS);
        }
    }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.