Copilot commented on code in PR #39:
URL: https://github.com/apache/pulsar-connectors/pull/39#discussion_r3553871949
##########
jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java:
##########
@@ -293,102 +293,114 @@ private void flush() {
}
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.error("Got exception {} after {} ms, failing {}
messages",
+ e.getMessage(),
+ (System.nanoTime() - start) / 1000 / 1000,
+ swapList.size(),
+ e);
Review Comment:
This log line now runs for any Throwable (including Errors). Using
e.getMessage() can produce "null" (e.g., StackOverflowError often has no
message), which reduces observability. Log the throwable’s toString()/class
name instead, and consider using TimeUnit for the duration conversion to keep
the message useful for Errors too.
##########
jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java:
##########
@@ -448,6 +452,102 @@ public void testBatchModeContinueFlushing() throws
Exception {
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 freezes flushing, queues a large backlog, 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);
+ // Disable the time-based flush; the only drain is the one we trigger
explicitly.
+ conf.put("timeoutMs", 0);
+ // Unbounded queue so enqueuing the whole backlog never applies
back-pressure.
+ conf.put("maxQueueSize", -1);
+
+ 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);
+
+ // Freeze draining so every write just accumulates the backlog.
+ FieldUtils.writeField(sink, "isFlushing", new AtomicBoolean(true),
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;
+
+ for (int i = 0; i < recordCount; i++) {
+ sink.write(record);
+ }
Review Comment:
With batchSize=1, each sink.write(record) schedules a flush task (delay=0).
Looping 50k times enqueues 50k scheduled tasks which can slow CI and introduce
races with the explicit reflective flush(). To keep the test deterministic and
fast, enqueue records directly into the sink’s incomingList under its lock
(this test already uses reflection for private fields/methods).
--
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]
