Copilot commented on code in PR #17:
URL: https://github.com/apache/pulsar-connectors/pull/17#discussion_r3553860497
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java:
##########
@@ -172,6 +177,10 @@ private static List<String> getListFromConfig(String
jdbcSinkConfig) {
@Override
public void close() throws Exception {
state.set(State.CLOSED);
+ // Fail any records still in the queue
+ List<Record<T>> remaining = new ArrayList<>();
+ incomingList.drainTo(remaining);
+ remaining.forEach(Record::fail);
Review Comment:
`close()` drains the entire queue into an `ArrayList` before failing
records. With `maxQueueSize = -1` (unbounded/legacy) or a very large queue,
this can allocate a massive list and potentially OOM during shutdown. Prefer
failing records incrementally (e.g., poll loop) to avoid materializing all
queued records at once.
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java:
##########
@@ -211,31 +220,30 @@ public void write(Record<T> record) throws Exception {
record.fail();
return;
}
- int number = 0;
- boolean shouldFail = false;
- boolean shouldLogQueueFull = false;
- int queueSizeSnapshot = 0;
- synchronized (incomingList) {
- if (maxQueueSize > 0 && incomingList.size() >= maxQueueSize) {
- if (!queueFullLogged) {
- queueFullLogged = true;
- shouldLogQueueFull = true;
- queueSizeSnapshot = incomingList.size();
- }
- shouldFail = true;
- } else {
- incomingList.add(record);
- number = incomingList.size();
- }
+
+ // Block until space is available or timeout expires.
+ // Blocking the Pulsar IO thread IS the back-pressure: it stops
+ // message consumption for this sink's subscription.
+ boolean accepted;
+ try {
+ accepted = incomingList.offer(record, 1, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ record.fail();
+ return;
}
- if (shouldFail) {
- if (shouldLogQueueFull) {
- log.warn("Internal queue is full ({} >= {}), failing records
to apply back-pressure",
- queueSizeSnapshot, maxQueueSize);
+
+ 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) {
Review Comment:
`write()` can enqueue a record after the sink transitions away from OPEN
(e.g., `close()` sets CLOSED while `offer()` is blocked/waiting). Because
`flush()` returns immediately when CLOSED, a record accepted after shutdown may
be left in the internal queue and never acked/failed. Re-check state after a
successful `offer()` and best-effort remove+fail if the sink is no longer OPEN.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
