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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
