darinspivey commented on code in PR #39:
URL: https://github.com/apache/pulsar-connectors/pull/39#discussion_r3554243360
##########
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:
Agreed, fixed in a2f4819. The backlog is now seeded directly onto
`incomingList` under its lock, so no flush tasks are scheduled at all. Since
`timeoutMs=0` also means no periodic flush, the drain is synchronous and
deterministic -- I dropped the `isFlushing` freeze and the Awaitility wait
along with it. The test still throws StackOverflowError against the old
recursive flush() and now runs in about a second.
--
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]
