Failure

Each error caught in a step gives a Failure object, sent to the failures PCollection.

Java

public class Failure implements Serializable {
    private final String pipelineStep;   // Name of the step
    private final String inputElement;   // Input element, as a string (toString)
    private final Throwable exception;   // Exception raised by the step
    private final String originElement;  // Element that entered the flow, null if not tracked (see Origin element)
    private final Instant timestamp;     // When the failure was created (since 1.3.0)

    public static <T> Failure from(String pipelineStep, WithFailures.ExceptionElement<T> exceptionElement);
    public static <T> Failure from(String pipelineStep, T element, Throwable exception);

    // Computed from the exception (since 1.3.0), e.g. to write the failure to a dead letter queue.
    public String getExceptionType();      // Class name, the original one for a SerializableThrowable
    public String getExceptionMessage();
    public String getStackTrace();         // Causes included

    // Encoded elements with their coder, null if not encoded (since 1.4.0, see withEncodedElements).
    public byte[] getInputElementBytes();
    public String getInputElementCoder();
    public byte[] getOriginElementBytes();
    public String getOriginElementCoder();

    // Getters...
}

To write the failures to BigQuery or to add them to a Beam ErrorHandler, see Write the failures and Beam native error handling.

If the exception can’t be serialized, it’s replaced by a SerializableThrowable keeping the original class name, message, stack trace and causes (see Failure handling guarantees).

Python

@dataclass
class Failure:
    pipeline_step: str       # Name of the step
    input_element: str       # Input element, as a string (JSON for a dict)
    exception: Exception     # Exception raised by the step (always picklable)
    stack_trace: str = ''    # Stack trace, as a string
    origin_element: str | None = None  # Element that entered the flow (see Origin element)
    timestamp: datetime | None = None  # When the failure was created, UTC (since 1.3.0)

    # Computed from the exception (since 1.3.0).
    exception_type: str                # Qualified name, the original one for a SerializableException
    exception_message: str | None

    def to_dict(self) -> dict: ...     # See Write the failures
    def to_bad_record(self): ...       # See Beam native error handling

    @classmethod
    def from_exception(cls, pipeline_step: str, element: Any, exception: Exception) -> 'Failure': ...

If the exception can’t be pickled, it’s replaced by a SerializableException keeping the original type and message. The stack trace is kept as a string because a pickled exception loses its traceback.

Input element format

The input element is stored as a string:

• Java: the toString() of the element.
• Python: the JSON string of a dict (non JSON types like datetime or bytes converted with str), the str() of any other element.

To control the format, give a function to the composer. It applies to all the next steps, and it’s the way to go for the classes you don’t own (Avro, Protobuf, third party classes): e.g. JSON, a format masking sensitive data, or an identifier.

Java

CollectionComposer.of(orders)
        .withInputElementToString(element -> JsonUtil.serialize(element))   // since 1.4.0
        .apply("Parse", ...)
        .apply("Validate", ...)
        .getResult();

Kotlin

CollectionComposer.of(orders)
    .withInputElementToString { element -> JsonUtil.serialize(element) }
    .mapFn("Parse", { order -> parse(order) })
    .result

Python

result = (CollectionComposer.of(orders)
          .with_input_element_to_string(lambda element: json.dumps(dataclasses.asdict(element)))   # since 1.4.0
          .map('Parse', parse)
          .filter('Validate', is_valid))

It also applies to the apply steps. In your own DoFn, give it to Failure.from_exception(step, element, err, element_to_string).

The function is evaluated only when a failure occurs. It never breaks the job: if it fails or returns null (None), the element is converted with the default conversion.

You can also implement the string representation of your own objects:

Java

public class Team implements Serializable {
    // Fields...

    @Override
    public String toString() {
        // Jackson is already a dependency of Beam.
        return JsonUtil.serialize(this);
    }
}

Python

@dataclass
class Team:
    name: str

    def __str__(self) -> str:
        return json.dumps(dataclasses.asdict(self))

Caution

The input element ends up in your dead letter queue: avoid writing sensitive data in the string representation of your objects.

Encoded elements

The string is enough to read, search and debug, not to replay an element exactly (republish the Pub/Sub or Kafka message as it entered, binary formats like Avro or Protobuf). With withEncodedElements, the failures also keep the input element, and the origin element, encoded with their coder:

Java

final WithFailures.Result<PCollection<Order>, Failure> result = CollectionComposer.of(messages)
        .withEncodedElements()   // since 1.4.0
        .apply("Parse", MapElementFn.into(TypeDescriptor.of(Order.class)).via(OrderParser::parse))
        .getResult();

// Replay: decode the element with its coder.
final Message message = CoderUtils.decodeFromByteArray(messageCoder, failure.getInputElementBytes());

Python

result = (CollectionComposer.of(messages)
          .with_encoded_elements()   # since 1.4.0
          .map('Parse', parse))

# Replay: decode the element with the coder of its type.
message = beam.coders.registry.get_coder(Message).decode(failure.input_element_bytes)

The coder is the registry coder of the element type of the PCollection consumed by the step. Beam Python often infers the types (StrUtf8Coder, TupleCoder, the schema coder of a NamedTuple…). Without type hints, the generic coder is used: it pickles custom objects, so these bytes are Python specific and depend on the class definitions at decode time. to_dict_with_encoded_elements() and FAILURE_BIGQUERY_SCHEMA_WITH_ENCODED_ELEMENTS give the bytes for BigQuery.

• The coder of the step: the element is encoded with the coder of the PCollection consumed by the step, the one the pipeline already uses for these elements. It works for every type Beam can carry: Avro, Protobuf, Row, KV, custom coders. The coder is also kept as a string (getInputElementCoder()).
• Only when a failure occurs, and it never breaks the job: if the element can’t be encoded, the failure has no bytes.
• Opt-in: the bytes make the dead letter queue bigger and keep the element in clear.

Limits

• Avro schema evolution: AvroCoder encodes without the schema. If the schema changes between the failure and the replay, prefer a JSON string with withInputElementToString for long term replays. Protobuf doesn’t have this issue.
• The coder is taken when the step is built: set the coders (setCoder) before applying the next steps.
• Failures built manually (custom exceptionsVia, own DoFn outside the composer): use failure.withEncodedInputElement(element, coder).