Copilot commented on code in PR #17:
URL: https://github.com/apache/pulsar-connectors/pull/17#discussion_r3553742043
##########
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() now unconditionally drains incomingList, but incomingList is
initialized late in open(). If open() fails before the queue is created and the
framework still calls close(), this will throw a NullPointerException and can
mask the original failure / skip cleanup. Guard the drain with a null check.
##########
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();
Review Comment:
write() blocks in incomingList.offer(...). If the sink transitions to
FAILED/CLOSED while the write thread is blocked, offer() can still succeed
after close() drains the queue, and the record may remain enqueued while
flush() becomes a no-op (state=CLOSED) or the executor is shut down. This can
leave the record neither acked nor failed. Re-check state after a successful
offer and, if no longer OPEN, remove the record (best-effort) and fail it
without scheduling a flush.
--
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]
