{"record":{"id":"4d640eb5f9fa2e1f","repo":"apache/beam","slug":"error-occurred-when-consuming-changes-from-database","errorCode":null,"errorMessage":"Error occurred when consuming changes from Database. ","messagePattern":"Error occurred when consuming changes from Database\\. ","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"critical","filePath":"sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java","lineNumber":313,"sourceCode":"          LOG.debug(\"****************** RECEIVED SOURCE AS JSON: {}\", json);\n\n          Instant recordInstant = debeziumRecordInstant(record);\n          receiver.outputWithTimestamp(json, recordInstant);\n        }\n        task.commit();\n\n        // Persist the offset after every successful commit so the pipeline can resume\n        // from this position on restart.\n        OffsetRetainer retainer = spec.getOffsetRetainer();\n        @SuppressWarnings(\"unchecked\")\n        Map<String, Object> committedOffset =\n            (Map<String, Object>) tracker.currentRestriction().offset;\n        if (retainer != null && committedOffset != null) {\n          retainer.saveOffset(committedOffset);\n        }\n      }\n    } catch (Exception ex) {\n      throw new RuntimeException(\"Error occurred when consuming changes from Database. \", ex);\n    } finally {\n      reset();\n\n      LOG.debug(\"------- Stopping SourceTask\");\n      task.stop();\n    }\n\n    if (spec.getMaxTimeToRun() != null && spec.getMaxTimeToRun() > 0) {\n      long elapsedTime = System.currentTimeMillis() - startTime.getMillis();\n      if (elapsedTime >= spec.getMaxTimeToRun()) {\n        return ProcessContinuation.stop();\n      }\n    }\n    return ProcessContinuation.resume()\n        .withResumeDelay(org.joda.time.Duration.millis(remainingTimeout.toMillis()));\n  }\n\n  public String getHashCode() {","sourceCodeStart":295,"sourceCodeEnd":331,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java#L295-L331","documentation":"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.","triggerScenarios":"Any failure inside the polling loop: connector task errors, deserialization failures, database connection drops, offset commit problems — anything thrown while consuming changes.","commonSituations":"Database failover or connection reset mid-stream, invalid record payloads that fail Connect deserialization, interrupted workers, bug in a Debezium converter.","solutions":["Inspect the cause chain ('Caused by') for the real failure and fix that root issue.","Check database connectivity/credentials and re-run; transient connection drops resolve on retry.","Use pipeline-level retry/DR options to restart the region from the retained offset.","Upgrade Debezium/Beam if the cause points to a converter bug."],"exampleFix":"// no caller-side code fix; diagnose via:\n// RuntimeException: Error occurred when consuming changes from Database.\n//   Caused by: org.apache.kafka.connect.errors.ConnectException: ... <- fix this root cause","handlingStrategy":"retry","validationCode":"// pre-flight: verify DB connectivity and connector health before launching\ntry (Connection c = DriverManager.getConnection(jdbcUrl, user, pass)) {\n  if (!c.isValid(5)) throw new IllegalStateException(\"DB unreachable\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  pipeline.run().waitUntilFinish();\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().startsWith(\"Error occurred when consuming changes from Database\")) {\n    Throwable root = e.getCause(); // diagnose root cause, then retry from retained offset\n  }\n}","preventionTips":["Monitor DB connection stability; use resilient endpoints/proxies","Keep the offset file intact so retries resume, not restart","Keep Debezium/Beam versions current to avoid converter bugs","Alert on pipeline failures and auto-retry transient causes"],"tags":["java","debezium","database","cdc","apache-beam"],"backgroundTag":"database-query-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}