{"record":{"id":"622d550451c082a9","repo":"apache/beam","slug":"failed-to-ingest-message","errorCode":null,"errorMessage":"Failed to ingest message","messagePattern":"Failed to ingest message","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java","lineNumber":960,"sourceCode":"        HL7v2Message msg = context.element();\n        // all fields but data and labels should be null for ingest.\n        Message model = new Message();\n        model.setData(msg.getData());\n        model.setLabels(msg.getLabels());\n        switch (writeMethod) {\n          case BATCH_IMPORT:\n            // TODO: add support for HL7v2 import.\n            throw new UnsupportedOperationException(\"The batch import API is not supported yet\");\n          case INGEST:\n          default:\n            try {\n              long requestTimestamp = Instant.now().getMillis();\n              client.ingestHL7v2Message(hl7v2Store.get(), model);\n              successfulHL7v2MessageWrites.inc();\n              messageIngestLatencyMs.update(Instant.now().getMillis() - requestTimestamp);\n            } catch (Exception e) {\n              failedMessageWrites.inc();\n              LOG.warn(\"Failed to ingest message\", e);\n              HealthcareIOError<HL7v2Message> err = HealthcareIOError.of(msg, e);\n              LOG.warn(\"{} {}\", err.getErrorMessage(), err.getStackTrace());\n              context.output(err);\n            }\n        }\n      }\n    }\n  }\n}\n","sourceCodeStart":942,"sourceCodeEnd":970,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java#L942-L970","documentation":"An HL7v2 message failed to be ingested into the Google Cloud Healthcare HL7v2 store via client.ingestHL7v2Message. The DoFn catches the exception, increments failedMessageWrites, logs it with stack trace, and emits a HealthcareIOError to the error output so the pipeline continues rather than crashing.","triggerScenarios":"Any Exception from ingestHL7v2Message during processElement: HL7v2 store not found, invalid/malformed HL7v2 message, permission denied, quota exceeded, or transient API errors.","commonSituations":"Feeding messages whose hospital/dataset/store path is wrong, malformed HL7v2 payloads (bad encodings, segment structure), service account missing healthcare.hl7V2Messages.ingest permission, Healthcare API downtime.","solutions":["Consume the error output PCollection (tagged errors) to inspect HealthcareIOError details and dead-letter or replay failed messages","Verify the HL7v2 store path (project/dataset/store) configured via HL7v2Store matches your environment","Validate HL7v2 message format before writing (e.g., with MLLP/HL7 parsers)","Grant roles/healthcare.hl7V2Ingest on the store to the pipeline service account"],"exampleFix":"// before\ntry { client.ingestHL7v2Message(hl7v2Store.get(), model); }\ncatch (Exception e) { failedMessageWrites.inc(); LOG.warn(\"Failed to ingest message\", e); }\n// after\ntry { client.ingestHL7v2Message(hl7v2Store.get(), model); }\ncatch (StatusRuntimeException e) {\n  failedMessageWrites.inc();\n  LOG.warn(\"Failed to ingest message, status={}\", e.getStatus(), e);\n  context.output(HealthcareIOError.of(msg, e));\n}","handlingStrategy":"try-catch","validationCode":"// pre-validate before ingesting\nif (!isValidHl7v2(model)) throw new IllegalArgumentException(\"malformed HL7v2 payload\");\nif (!storeExists(hl7v2Store.get())) throw new IllegalStateException(\"HL7v2 store missing: \" + hl7v2Store.get());","typeGuard":null,"tryCatchPattern":"try {\n  client.ingestHL7v2Message(hl7v2Store.get(), model);\n} catch (StatusRuntimeException e) {\n  LOG.warn(\"ingest failed, status={}\", e.getStatus(), e);\n  context.output(HealthcareIOError.of(msg, e));\n} catch (Exception e) {\n  LOG.warn(\"ingest failed unexpectedly\", e);\n  context.output(HealthcareIOError.of(msg, e));\n}","preventionTips":["Always sink the HealthcareIOError error output to a durable dead-letter store","Validate HL7v2 message structure before ingest","Verify store path (project/dataset/store) and healthcare.hl7V2Ingest IAM before launching","Monitor failedMessageWrites counter and alert on spikes"],"tags":["gcp","healthcare","hl7v2","dead-letter"],"backgroundTag":"api-request-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}