apache/beam · error

Failed to ingest message

Error message

Failed to ingest message

What it means

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.

Solutions

1. Consume the error output PCollection (tagged errors) to inspect HealthcareIOError details and dead-letter or replay failed messages
2. Verify the HL7v2 store path (project/dataset/store) configured via HL7v2Store matches your environment
3. Validate HL7v2 message format before writing (e.g., with MLLP/HL7 parsers)
4. Grant roles/healthcare.hl7V2Ingest on the store to the pipeline service account

Example fix

// before
try { client.ingestHL7v2Message(hl7v2Store.get(), model); }
catch (Exception e) { failedMessageWrites.inc(); LOG.warn("Failed to ingest message", e); }
// after
try { client.ingestHL7v2Message(hl7v2Store.get(), model); }
catch (StatusRuntimeException e) {
  failedMessageWrites.inc();
  LOG.warn("Failed to ingest message, status={}", e.getStatus(), e);
  context.output(HealthcareIOError.of(msg, e));
}

// pre-validate before ingesting
if (!isValidHl7v2(model)) throw new IllegalArgumentException("malformed HL7v2 payload");
if (!storeExists(hl7v2Store.get())) throw new IllegalStateException("HL7v2 store missing: " + hl7v2Store.get());

try {
  client.ingestHL7v2Message(hl7v2Store.get(), model);
} catch (StatusRuntimeException e) {
  LOG.warn("ingest failed, status={}", e.getStatus(), e);
  context.output(HealthcareIOError.of(msg, e));
} catch (Exception e) {
  LOG.warn("ingest failed unexpectedly", e);
  context.output(HealthcareIOError.of(msg, e));
}

When it happens

Trigger: Any Exception from ingestHL7v2Message during processElement: HL7v2 store not found, invalid/malformed HL7v2 message, permission denied, quota exceeded, or transient API errors.

Common situations: 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.

Understand the failure class

Background: "API request failed": what wrapped HTTP errors from external APIs mean and how to find the real cause — this error's family across 29 libraries.

Related errors

• Failed to import with error. Moving to deadletter path
• {} {}
• DicomSearch failed with status
• Error fetching HL7v2 message with ID
• Error while parsing the DataChangeRecord

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

Appendix: source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java:960

        HL7v2Message msg = context.element();
        // all fields but data and labels should be null for ingest.
        Message model = new Message();
        model.setData(msg.getData());
        model.setLabels(msg.getLabels());
        switch (writeMethod) {
          case BATCH_IMPORT:
            // TODO: add support for HL7v2 import.
            throw new UnsupportedOperationException("The batch import API is not supported yet");
          case INGEST:
          default:
            try {
              long requestTimestamp = Instant.now().getMillis();
              client.ingestHL7v2Message(hl7v2Store.get(), model);
              successfulHL7v2MessageWrites.inc();
              messageIngestLatencyMs.update(Instant.now().getMillis() - requestTimestamp);
            } catch (Exception e) {
              failedMessageWrites.inc();
              LOG.warn("Failed to ingest message", e);
              HealthcareIOError<HL7v2Message> err = HealthcareIOError.of(msg, e);
              LOG.warn("{} {}", err.getErrorMessage(), err.getStackTrace());
              context.output(err);
            }
        }
      }
    }
  }
}

View on GitHub (pinned to 12126d8942)