apache/beam · critical · RuntimeException

Error occurred when consuming changes from Database.

Error message

Error occurred when consuming changes from Database. 

What it means

The DoFn.process method of KafkaSourceConsumerFn wraps its entire consume loop in a try block; any Exception thrown while reading CDC events from the database is rethrown as this RuntimeException with the original as the cause. The finally block still resets state and stops the SourceTask.

Source

Thrown at sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:313

          LOG.debug("****************** RECEIVED SOURCE AS JSON: {}", json);

          Instant recordInstant = debeziumRecordInstant(record);
          receiver.outputWithTimestamp(json, recordInstant);
        }
        task.commit();

        // Persist the offset after every successful commit so the pipeline can resume
        // from this position on restart.
        OffsetRetainer retainer = spec.getOffsetRetainer();
        @SuppressWarnings("unchecked")
        Map<String, Object> committedOffset =
            (Map<String, Object>) tracker.currentRestriction().offset;
        if (retainer != null && committedOffset != null) {
          retainer.saveOffset(committedOffset);
        }
      }
    } catch (Exception ex) {
      throw new RuntimeException("Error occurred when consuming changes from Database. ", ex);
    } finally {
      reset();

      LOG.debug("------- Stopping SourceTask");
      task.stop();
    }

    if (spec.getMaxTimeToRun() != null && spec.getMaxTimeToRun() > 0) {
      long elapsedTime = System.currentTimeMillis() - startTime.getMillis();
      if (elapsedTime >= spec.getMaxTimeToRun()) {
        return ProcessContinuation.stop();
      }
    }
    return ProcessContinuation.resume()
        .withResumeDelay(org.joda.time.Duration.millis(remainingTimeout.toMillis()));
  }

  public String getHashCode() {

View on GitHub (pinned to 12126d8942)

Solutions

1. Inspect the cause chain ('Caused by') for the real failure and fix that root issue.
2. Check database connectivity/credentials and re-run; transient connection drops resolve on retry.
3. Use pipeline-level retry/DR options to restart the region from the retained offset.
4. Upgrade Debezium/Beam if the cause points to a converter bug.

Example fix

// no caller-side code fix; diagnose via:
// RuntimeException: Error occurred when consuming changes from Database.
//   Caused by: org.apache.kafka.connect.errors.ConnectException: ... <- fix this root cause

// pre-flight: verify DB connectivity and connector health before launching
try (Connection c = DriverManager.getConnection(jdbcUrl, user, pass)) {
  if (!c.isValid(5)) throw new IllegalStateException("DB unreachable");
}

try {
  pipeline.run().waitUntilFinish();
} catch (RuntimeException e) {
  if (e.getMessage() != null && e.getMessage().startsWith("Error occurred when consuming changes from Database")) {
    Throwable root = e.getCause(); // diagnose root cause, then retry from retained offset
  }
}

When it happens

Trigger: Any failure inside the polling loop: connector task errors, deserialization failures, database connection drops, offset commit problems — anything thrown while consuming changes.

Common situations: Database failover or connection reset mid-stream, invalid record payloads that fail Connect deserialization, interrupted workers, bug in a Debezium converter.

Understand the failure class

Background: Database query failed: Internal Server Error 500s wrapping SQL, Prisma, and connection failures — what to check first — this error's family across 16 libraries.

Related errors

• could not fetch database schema
• Unable to resolve class %s to use as Debezium connector.
• Cannot create enum from value!
• Unsupported connector ''. Supported connectors are:
• host must not be empty.

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/4d640eb5f9fa2e1f. Report an issue: GitHub.