You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pulsar.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/04/01 03:54:33 UTC
Re: [PR] [fix][io] JDBC sink: prevent OOM from unbounded queue on connection failure [pulsar-connectors]
Copilot commented on code in PR #9:
URL: https://github.com/apache/pulsar-connectors/pull/9#discussion_r3019608721
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java:
##########
@@ -239,6 +270,9 @@ protected enum MutationType {
private void flush() {
+ if (state.get() == State.CLOSED) {
+ return;
+ }
if (incomingList.size() > 0 && isFlushing.compareAndSet(false, true)) {
Review Comment:
flush() reads incomingList.size() outside of synchronized(incomingList) (line 276, and the debug log at 371). Since LinkedList is not thread-safe, this is a data race and can lead to flush being skipped even when items are present (or to inconsistent queue size observations). Move the emptiness/size check inside the existing synchronized block (or switch to a concurrent collection) so flush triggering is based on a consistent view.
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java:
##########
@@ -114,12 +119,20 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
int timeoutMs = jdbcSinkConfig.getTimeoutMs();
batchSize = jdbcSinkConfig.getBatchSize();
+ maxQueueSize = jdbcSinkConfig.getMaxQueueSize();
+ if (maxQueueSize == 0) {
+ // Auto-size: default to 10x batch size
+ maxQueueSize = batchSize > 0 ? batchSize * 10 : 10000;
Review Comment:
maxQueueSize auto-sizing uses `batchSize * 10` on ints. For very large configured batchSize values this can overflow to a negative number and unintentionally disable the limit (and reintroduce OOM risk). Use a safe multiply (e.g., long with cap at Integer.MAX_VALUE, or Math.multiplyExact with fallback) before assigning to maxQueueSize.
```suggestion
// Auto-size: default to 10x batch size (with overflow-safe calculation)
long calculated = batchSize > 0 ? (long) batchSize * 10L : 10000L;
if (calculated > Integer.MAX_VALUE) {
maxQueueSize = Integer.MAX_VALUE;
} else {
maxQueueSize = (int) calculated;
}
```
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcSinkConfig.java:
##########
@@ -130,6 +130,18 @@ public class JdbcSinkConfig implements Serializable {
)
private NullValueAction nullValueAction = NullValueAction.FAIL;
+ @FieldDoc(
+ required = false,
+ defaultValue = "0",
+ help = "Maximum number of records to buffer in the internal queue before applying back-pressure. "
+ + "When the queue is full, incoming records will be failed (negatively acknowledged) so that "
+ + "the Pulsar consumer can redeliver them later. This prevents out-of-memory errors when the "
+ + "database connection is slow or broken. "
+ + "A value of 0 (default) auto-sizes to batchSize * 10. "
+ + "A value of -1 disables the limit (unbounded, legacy behavior)."
+ )
+ private int maxQueueSize = 0;
Review Comment:
This PR introduces a new default behavior: previously the in-memory queue was effectively unbounded; now maxQueueSize defaults to 0 and is interpreted as a bounded auto-size (batchSize * 10). This can cause records to be failed/redelivered under sustained backlog where previously they would queue. If the intent is to preserve legacy behavior by default, consider defaulting to -1 or treating 0 as unbounded; otherwise, please ensure the PR description/release notes reflect the behavioral change (the PR description currently states that `maxQueueSize=0` preserves legacy unbounded behavior).
```suggestion
defaultValue = "-1",
help = "Maximum number of records to buffer in the internal queue before applying back-pressure. "
+ "When the queue is full, incoming records will be failed (negatively acknowledged) so that "
+ "the Pulsar consumer can redeliver them later. This prevents out-of-memory errors when the "
+ "database connection is slow or broken. "
+ "A value of 0 auto-sizes to batchSize * 10. "
+ "A value of -1 (default) disables the limit (unbounded, legacy behavior)."
)
private int maxQueueSize = -1;
```
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java:
##########
@@ -188,8 +205,22 @@ public void close() throws Exception {
@Override
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();
}
Review Comment:
write() calls record.fail() while holding the incomingList monitor. Record callbacks can execute user/framework code and potentially block, which can stall concurrent write()/flush() and increase back-pressure unexpectedly. Consider capturing a boolean "shouldFail" inside the synchronized block and invoking record.fail() after releasing the lock.
```suggestion
boolean shouldFail = false;
boolean shouldLogQueueFull = false;
int queueSizeAtFailure = -1;
synchronized (incomingList) {
if (maxQueueSize > 0 && incomingList.size() >= maxQueueSize) {
if (!queueFullLogged) {
queueFullLogged = true;
shouldLogQueueFull = true;
queueSizeAtFailure = incomingList.size();
}
shouldFail = true;
} else {
incomingList.add(record);
number = incomingList.size();
}
}
if (shouldFail) {
if (shouldLogQueueFull) {
log.warn("Internal queue is full ({} >= {}), failing records to apply back-pressure",
queueSizeAtFailure, maxQueueSize);
}
record.fail();
return;
}
```
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcSinkConfig.java:
##########
@@ -130,6 +130,18 @@ public class JdbcSinkConfig implements Serializable {
)
private NullValueAction nullValueAction = NullValueAction.FAIL;
+ @FieldDoc(
+ required = false,
+ defaultValue = "0",
+ help = "Maximum number of records to buffer in the internal queue before applying back-pressure. "
+ + "When the queue is full, incoming records will be failed (negatively acknowledged) so that "
+ + "the Pulsar consumer can redeliver them later. This prevents out-of-memory errors when the "
+ + "database connection is slow or broken. "
+ + "A value of 0 (default) auto-sizes to batchSize * 10. "
+ + "A value of -1 disables the limit (unbounded, legacy behavior)."
+ )
+ private int maxQueueSize = 0;
Review Comment:
The FieldDoc says "-1 disables the limit", but the implementation treats any negative value as unbounded (since only `maxQueueSize > 0` is enforced). To avoid surprising behavior from typos (e.g., -2), consider validating maxQueueSize in JdbcSinkConfig.validate() to allow only -1, 0, or a positive value (and fail fast otherwise).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org