You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "darinspivey (via GitHub)" <gi...@apache.org> on 2026/07/08 20:08:16 UTC
[PR] [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under sustained backlog [pulsar-connectors]
darinspivey opened a new pull request, #39:
URL: https://github.com/apache/pulsar-connectors/pull/39
### Problem
Under sustained catch-up (a backlog is present and the arrival rate is greater than the 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, the consumer blocked at its unacked cap, the whole Shared subscription stalled, and no logged error.
This reproduces in production JDBC sinks during backlog recovery and traffic spikes. It is present in v4.1.3, branch-4.1, and current `master`.
### 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.
### Note on diff size
Most of the diff is re-indentation from wrapping the body in `try/finally` and `do/while`. Reviewing with `git diff -w` shows the real delta is about 30 lines.
### Testing
Adds `SqliteJdbcSinkTest.testFlushDrainsLargeBacklogWithoutStackOverflow`, which freezes flushing, queues a large backlog, then drives a single `flush()`. With `batchSize=1` the drain performs one round per record, so the backlog size equals the old recursion depth. The test throws `StackOverflowError` on the old recursive code and passes on the iterative version, and asserts `fatal()` is never invoked. The full `jdbc` test suite, `build`, `rat`, and `spotlessCheck` pass locally.
Fixes #38
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under sustained backlog [pulsar-connectors]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under sustained backlog [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio merged PR #39:
URL: https://github.com/apache/pulsar-connectors/pull/39
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under sustained backlog [pulsar-connectors]
Posted by "darinspivey (via GitHub)" <gi...@apache.org>.
darinspivey commented on code in PR #39:
URL: https://github.com/apache/pulsar-connectors/pull/39#discussion_r3554241902
##########
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:
Good catch, fixed in a2f4819. A StackOverflowError has no message, so this would have logged "Got exception null" -- exactly the kind of blind spot this PR is meant to remove. Now logs `e.toString()`, uses `TimeUnit.NANOSECONDS.toMillis()` for the duration, and says "throwable" since Errors are caught here.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under sustained backlog [pulsar-connectors]
Posted by "darinspivey (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under sustained backlog [pulsar-connectors]
Posted by "darinspivey (via GitHub)" <gi...@apache.org>.
darinspivey commented on PR #39:
URL: https://github.com/apache/pulsar-connectors/pull/39#issuecomment-4928791394
FWIW, I did a custom build with this fix and it unblocked this issue we've seen in production. Always fun to "test it live" :)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org