This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 71bdfe69 [fix][io] jdbc: drain flush() iteratively to avoid
StackOverflowError under sustained backlog (#39)
71bdfe69 is described below
commit 71bdfe69a50a08fdc329f443cbbd06ee51aba59d
Author: Darin Spivey <[email protected]>
AuthorDate: Fri Jul 10 10:53:55 2026 -0400
[fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under
sustained backlog (#39)
* [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError
under sustained backlog
Problem. Under sustained catch-up (backlog present, arrival rate greater
than
flush rate) JdbcAbstractSink.flush() recurses via `if (needAnotherRound)
flush();`
and never unwinds, eventually throwing StackOverflowError. The handler is
catch (Exception), not Throwable, so the Error bypasses the fatal() path
added in
PIP-297 (#25195). The ScheduledThreadPoolExecutor swallows it and cancels
the
periodic flush, and the sink hangs silently: 1/1 Ready, zero writes and
acks,
consumer blocked at the unacked cap, whole Shared subscription stalled, no
logged
error.
Fix.
1. Convert the recursion to a `do { ... } while (needAnotherRound)` loop.
2. Widen catch (Exception) to catch (Throwable) and change fatal(Exception)
to
fatal(Throwable) so an Error still triggers the instance restart instead
of
vanishing.
3. Release isFlushing in a single finally, which also fixes a latent
stuck-flag
path.
Most of the diff is re-indentation from wrapping the body in try/finally and
do/while. `git diff -w` shows the real delta is about 30 lines.
Testing. Adds a regression test in the sqlite module that queues a large
backlog
and drives a single flush(), which StackOverflows on the old code and
completes
on the fixed code.
Fixes #38.
* [fix][io] jdbc: address review feedback on the flush() fix
Log the throwable's toString() instead of getMessage(). An Error such as
StackOverflowError carries no message, so the line rendered as "Got
exception null" and hid what actually failed. Also convert the elapsed
time with TimeUnit.NANOSECONDS.toMillis(), and say "throwable" rather
than "exception" now that Errors are caught here.
Seed the regression test's backlog directly onto the sink's incomingList
instead of calling write() 50k times. Each write() scheduled a flush
task (batchSize=1), which was slow on CI and raced with the explicit
flush(). With no scheduled tasks the drain is synchronous and
deterministic, so the isFlushing freeze and the Awaitility wait are no
longer needed.
Ref: apache/pulsar-connectors#38
---------
Co-authored-by: David Kjerrumgaard
<[email protected]>
---
.../apache/pulsar/io/jdbc/JdbcAbstractSink.java | 189 +++++++++++----------
.../apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java | 104 ++++++++++++
2 files changed, 206 insertions(+), 87 deletions(-)
diff --git
a/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
b/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
index 3f2477a3..7fc8baa4 100644
--- a/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
+++ b/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
@@ -293,102 +293,117 @@ public abstract class JdbcAbstractSink<T> implements
Sink<T> {
}
return;
}
- boolean needAnotherRound;
- final Deque<Record<T>> swapList = new LinkedList<>();
-
- synchronized (incomingList) {
- if (incomingList.isEmpty()) {
- isFlushing.set(false);
- return;
- }
- if (log.isDebugEnabled()) {
- log.debug("Starting flush, queue size: {}",
incomingList.size());
- }
- final int actualBatchSize = batchSize > 0 ?
Math.min(incomingList.size(), batchSize) :
- incomingList.size();
+ // Drain queued batches iteratively. This was previously a
tail-recursive call
+ // (flush() calling itself when another full batch was queued), which
under sustained
+ // catch-up load never unwinds the stack and eventually throws
StackOverflowError.
+ try {
+ boolean needAnotherRound = false;
+ do {
+ final Deque<Record<T>> swapList = new LinkedList<>();
- for (int i = 0; i < actualBatchSize; i++) {
- swapList.add(incomingList.removeFirst());
- }
- needAnotherRound = batchSize > 0 && !incomingList.isEmpty() &&
incomingList.size() >= batchSize;
- }
- long start = System.nanoTime();
+ synchronized (incomingList) {
+ if (incomingList.isEmpty()) {
+ return;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Starting flush, queue size: {}",
incomingList.size());
+ }
+ final int actualBatchSize = batchSize > 0 ?
Math.min(incomingList.size(), batchSize) :
+ incomingList.size();
- int count = 0;
- try {
- ensureConnection();
-
- PreparedStatement currentBatch = null;
- final List<Mutation> mutations = swapList
- .stream()
- .map(this::createMutation)
- .collect(Collectors.toList());
- // bind each record value
- PreparedStatement statement;
- for (Mutation mutation : mutations) {
- switch (mutation.getType()) {
- case DELETE:
- statement = deleteStatement;
- break;
- case UPDATE:
- statement = updateStatement;
- break;
- case INSERT:
- statement = insertStatement;
- break;
- case UPSERT:
- statement = upsertStatement;
- break;
- default:
- String msg = String.format(
- "Unsupported action %s, can be one of %s,
or not set which indicate %s",
- mutation.getType(),
Arrays.toString(MutationType.values()), MutationType.INSERT);
- throw new IllegalArgumentException(msg);
+ for (int i = 0; i < actualBatchSize; i++) {
+ swapList.add(incomingList.removeFirst());
}
- bindValue(statement, mutation);
- count += 1;
- if (jdbcSinkConfig.isUseJdbcBatch()) {
- if (currentBatch != null && statement != currentBatch)
{
- internalFlushBatch(swapList, currentBatch, count,
start);
- start = System.nanoTime();
+ needAnotherRound = batchSize > 0 && !incomingList.isEmpty()
+ && incomingList.size() >= batchSize;
+ }
+ long start = System.nanoTime();
+
+ int count = 0;
+ try {
+ ensureConnection();
+
+ PreparedStatement currentBatch = null;
+ final List<Mutation> mutations = swapList
+ .stream()
+ .map(this::createMutation)
+ .collect(Collectors.toList());
+ // bind each record value
+ PreparedStatement statement;
+ for (Mutation mutation : mutations) {
+ switch (mutation.getType()) {
+ case DELETE:
+ statement = deleteStatement;
+ break;
+ case UPDATE:
+ statement = updateStatement;
+ break;
+ case INSERT:
+ statement = insertStatement;
+ break;
+ case UPSERT:
+ statement = upsertStatement;
+ break;
+ default:
+ String msg = String.format(
+ "Unsupported action %s, can be one of
%s, or not set which indicate %s",
+ mutation.getType(),
Arrays.toString(MutationType.values()),
+ MutationType.INSERT);
+ throw new IllegalArgumentException(msg);
}
- statement.addBatch();
- currentBatch = statement;
- } else {
- statement.execute();
- if (!jdbcSinkConfig.isUseTransactions()) {
- swapList.removeFirst().ack();
+ bindValue(statement, mutation);
+ count += 1;
+ if (jdbcSinkConfig.isUseJdbcBatch()) {
+ if (currentBatch != null && statement !=
currentBatch) {
+ internalFlushBatch(swapList, currentBatch,
count, start);
+ start = System.nanoTime();
+ }
+ statement.addBatch();
+ currentBatch = statement;
+ } else {
+ statement.execute();
+ if (!jdbcSinkConfig.isUseTransactions()) {
+ swapList.removeFirst().ack();
+ }
}
}
- }
- if (jdbcSinkConfig.isUseJdbcBatch()) {
- internalFlushBatch(swapList, currentBatch, count, start);
- } else {
- internalFlush(swapList);
- }
- queueFullLogged = false;
- } catch (Exception e) {
- log.error("Got exception {} after {} ms, failing {} messages",
- e.getMessage(),
- (System.nanoTime() - start) / 1000 / 1000,
- swapList.size(),
- e);
- swapList.forEach(Record::fail);
- try {
- if (jdbcSinkConfig.isUseTransactions()) {
- connection.rollback();
+ if (jdbcSinkConfig.isUseJdbcBatch()) {
+ internalFlushBatch(swapList, currentBatch, count,
start);
+ } else {
+ internalFlush(swapList);
+ }
+ queueFullLogged = false;
+ } catch (Throwable e) {
+ // Catch Throwable (not just Exception) so an Error such
as StackOverflowError
+ // is surfaced to the framework via fatal() and the
instance is terminated and
+ // restarted, instead of the flush thread dying silently
while the pod keeps
+ // reporting healthy.
+ // Log toString() rather than getMessage(): an Error such
as
+ // StackOverflowError usually carries no message, which
would render as
+ // "null" and hide what actually failed.
+ log.error("Got throwable {} after {} ms, failing {}
messages",
+ e.toString(),
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
start),
+ swapList.size(),
+ e);
+ swapList.forEach(Record::fail);
+ try {
+ if (jdbcSinkConfig.isUseTransactions()) {
+ connection.rollback();
+ }
+ } catch (Exception ex) {
+ log.error("Failed to rollback transaction", ex);
}
- } catch (Exception ex) {
- log.error("Failed to rollback transaction", ex);
+ fatal(e);
+ needAnotherRound = false;
}
- fatal(e);
- }
-
+ } while (needAnotherRound);
+ } finally {
+ // Always release the flag, even if an Error escaped the inner
try, otherwise the
+ // sink would be permanently stuck in the flushing state.
isFlushing.set(false);
- if (needAnotherRound) {
- flush();
- }
+ }
}
private void ensureConnection() throws Exception {
@@ -489,7 +504,7 @@ public abstract class JdbcAbstractSink<T> implements
Sink<T> {
*
* @param e the fatal exception
*/
- private void fatal(Exception e) {
+ private void fatal(Throwable e) {
if (sinkContext != null && state.compareAndSet(State.OPEN,
State.FAILED)) {
log.error("Fatal error in JDBC sink, signaling framework for
shutdown", e);
if (scheduledFlushTask != null) {
diff --git
a/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java
b/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java
index c76436a1..74005475 100644
---
a/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java
+++
b/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java
@@ -22,13 +22,16 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
+import java.util.Deque;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -38,6 +41,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import lombok.AllArgsConstructor;
import lombok.Data;
@@ -45,6 +49,7 @@ import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.util.Utf8;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.reflect.MethodUtils;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.schema.GenericObject;
@@ -448,6 +453,105 @@ public class SqliteJdbcSinkTest {
futureByEntries2.get(1, TimeUnit.SECONDS);
}
+ /**
+ * Regression test for a StackOverflowError in {@code
JdbcAbstractSink.flush()}.
+ *
+ * <p>flush() used to drain "another round" by calling itself recursively
+ * ({@code if (needAnotherRound) { flush(); }}). Under sustained catch-up
a single
+ * flush() invocation has to drain many queued batches, so the recursion
depth grew
+ * with the backlog and eventually threw StackOverflowError. Because the
handler only
+ * caught Exception (not Throwable), that Error escaped: the scheduled
flush executor
+ * swallowed it and cancelled the periodic flush, and the sink hung
silently while
+ * still reporting healthy.
+ *
+ * <p>This test seeds a large backlog straight onto the sink's queue, then
lets a single
+ * flush() drain all of it. With {@code batchSize=1} the drain performs
one round per
+ * record, so the backlog size equals the old recursion depth. The insert
statement is
+ * replaced with a no-op mock (as in {@code
testFatalCalledOnFlushException}) so the drain
+ * exercises the flush control flow at high volume without real database
I/O. The
+ * iterative implementation completes and acks every record; the recursive
one
+ * overflowed the stack. fatal() must never be invoked.
+ */
+ @Test
+ public void testFlushDrainsLargeBacklogWithoutStackOverflow() throws
Exception {
+ jdbcSink.close();
+ jdbcSink = null;
+
+ // Large enough that the old one-frame-per-batch recursion overflows a
default
+ // thread stack with margin, even when the test runner uses a bigger
-Xss.
+ final int recordCount = 50_000;
+
+ Map<String, Object> conf = Maps.newHashMap();
+ conf.put("jdbcUrl", sqliteUtils.sqliteUri());
+ conf.put("tableName", tableName);
+ conf.put("key", "field3");
+ conf.put("nonKey", "field1,field2");
+ // One record per flush round, so the number of rounds equals the
backlog size.
+ conf.put("batchSize", 1);
+ // No periodic flush task is scheduled, so the only drain is the one
we invoke below.
+ conf.put("timeoutMs", 0);
+
+ SinkContext mockSinkContext = mock(SinkContext.class);
+ SqliteJdbcAutoSchemaSink sink = new SqliteJdbcAutoSchemaSink();
+ try {
+ sink.open(conf, mockSinkContext);
+
+ // Replace the insert statement with a no-op so a large drain
stays fast and
+ // deterministic; this test exercises flush()'s control flow, not
the database.
+ PreparedStatement noopStatement = mock(PreparedStatement.class);
+ FieldUtils.writeField(sink, "insertStatement", noopStatement,
true);
+
+ // The recursion depended on the queue depth, not on record
contents, so a
+ // single record enqueued many times is enough (and keeps the test
cheap).
+ AtomicLong acked = new AtomicLong();
+ AtomicLong failed = new AtomicLong();
+ AvroSchema<Foo> schema =
AvroSchema.of(SchemaDefinition.<Foo>builder()
+ .withPojo(Foo.class).withAlwaysAllowNull(true).build());
+ AutoConsumeSchema autoConsumeSchema = new AutoConsumeSchema();
+ autoConsumeSchema.setSchema(schema);
+ GenericAvroSchema genericAvroSchema = new
GenericAvroSchema(schema.getSchemaInfo());
+ byte[] bytes = schema.encode(new Foo("f1", "f2", 1));
+ Map<String, String> insertProps = Maps.newHashMap();
+ insertProps.put("ACTION", "INSERT");
+ Message<GenericRecord> message = mock(MessageImpl.class,
withSettings().stubOnly());
+
when(message.getValue()).thenReturn(genericAvroSchema.decode(bytes));
+ when(message.getProperties()).thenReturn(insertProps);
+ Record<? extends GenericObject> builtRecord =
PulsarRecord.<GenericRecord>builder()
+ .message(message)
+ .topicName("fake_topic_name")
+ .schema(autoConsumeSchema)
+ .ackFunction(acked::incrementAndGet)
+ .failFunction(failed::incrementAndGet)
+ .build();
+ @SuppressWarnings("unchecked")
+ Record<GenericObject> record = (Record<GenericObject>) builtRecord;
+
+ // Seed the backlog directly on the sink's queue. Going through
write() would
+ // schedule one flush task per record (batchSize=1), which is slow
and races with
+ // the explicit flush() below. This test is about how flush()
drains a deep queue,
+ // so seeding the queue keeps it fast and fully deterministic.
+ @SuppressWarnings("unchecked")
+ Deque<Record<GenericObject>> incomingList =
+ (Deque<Record<GenericObject>>) FieldUtils.readField(sink,
"incomingList", true);
+ synchronized (incomingList) {
+ for (int i = 0; i < recordCount; i++) {
+ incomingList.add(record);
+ }
+ }
+
+ // Nothing else can flush, so this single call must drain the
entire backlog
+ // synchronously: one round per record.
+ MethodUtils.invokeMethod(sink, true, "flush");
+
+ Assert.assertEquals(acked.get(), recordCount);
+ Assert.assertEquals(failed.get(), 0L);
+ // The Error path must not have been hit at all.
+ verify(mockSinkContext, never()).fatal(any(Throwable.class));
+ } finally {
+ sink.close();
+ }
+ }
+
@DataProvider(name = "useTransactions")
public Object[] useTransactions() {
return Arrays.asList(true, false).toArray();
