Why Asgarde

Beam recommends handling errors with dead letter queues: the errors are caught in the flow and sent, via side outputs, to a dedicated sink (a file, a table, a topic…) instead of failing the job.

Error handling with plain Beam

With plain Beam, every step needs its own error handling code, and all the failures must be gathered at the end:

Java

WithFailures.Result<PCollection<String>, Failure> result1 = input
        .apply("Map", MapElements
                .into(TypeDescriptors.strings())
                .via((String word) -> word + "Test")
                .exceptionsInto(TypeDescriptor.of(Failure.class))
                .exceptionsVia(exElt -> Failure.from("Map", exElt)));

WithFailures.Result<PCollection<String>, Failure> result2 = result1.output()
        .apply("FlatMap", FlatMapElements
                .into(TypeDescriptors.strings())
                .via((String line) -> Arrays.asList(line.split(" ")))
                .exceptionsInto(TypeDescriptor.of(Failure.class))
                .exceptionsVia(exElt -> Failure.from("FlatMap", exElt)));

final PCollectionTuple result3 = result2.output()
        .apply("Word count", ParDo.of(wordCountFn)
                .withOutputTags(wordCountFn.getOutputTag(), TupleTagList.of(wordCountFn.getFailuresTag())));

final PCollection<Integer> output = result3.get(wordCountFn.getOutputTag());
final PCollection<Failure> allFailures = PCollectionList
        .of(result1.failures())
        .and(result2.failures())
        .and(result3.get(wordCountFn.getFailuresTag()))
        .apply(Flatten.pCollections());

Plus, for each custom DoFn, the tuple tags and the try/catch block:

public class WordCountFn extends DoFn<String, Integer> {
    private final TupleTag<Integer> outputTag = new TupleTag<Integer>() {};
    private final TupleTag<Failure> failuresTag = new TupleTag<Failure>() {};

    @ProcessElement
    public void processElement(ProcessContext ctx) {
        try {
            ctx.output(1 / ctx.element().length());
        } catch (Exception e) {
            ctx.output(failuresTag, Failure.from("Word count", ctx.element(), e));
        }
    }
    // Getters for the tags...
}

Python

class MapToTeamWithCountry(beam.DoFn):
    def process(self, element):
        try:
            yield TeamInfo(name=element, country=team_countries[element], city='')
        except Exception as err:
            yield pvalue.TaggedOutput(FAILURES, Failure(pipeline_step='Map 1', input_element=element, exception=err))

# Same DoFn boilerplate for each step...

outputs_map1, failures_map1 = (input_teams
                               | 'Map to team with country' >> beam.ParDo(MapToTeamWithCountry())
                               .with_outputs(FAILURES, main='outputs'))

outputs_map2, failures_map2 = (outputs_map1
                               | 'Map to team with city' >> beam.ParDo(MapToTeamWithCity())
                               .with_outputs(FAILURES, main='outputs'))

outputs_filter, failures_filter = (outputs_map2
                                   | 'Filter France teams' >> beam.ParDo(FilterFranceTeams())
                                   .with_outputs(FAILURES, main='outputs'))

all_failures = (failures_map1, failures_map2, failures_filter) | 'All failures' >> beam.Flatten()

The problems with this approach:

• The fluent style is lost: the output and the failures must be handled for each step.
• The same technical code is repeated everywhere: exceptionsInto/exceptionsVia, tuple tags, try/catch blocks.
• All the failures must be concatenated manually at the end.
• The code is verbose and error-prone.

The same flow with Asgarde

Java

final WithFailures.Result<PCollection<Integer>, Failure> result = CollectionComposer.of(input)
        .apply("Map", MapElements.into(TypeDescriptors.strings()).via((String word) -> word + "Test"))
        .apply("FlatMap", FlatMapElements
                .into(TypeDescriptors.strings())
                .via((String line) -> Arrays.asList(line.split(" "))))
        .apply("Word count", MapElementFn.into(TypeDescriptors.integers()).via(word -> 1 / word.length()))
        .getResult();

Kotlin

val result: Result<PCollection<Int>, Failure> = CollectionComposer.of(input)
    .map("Map") { word -> word + "Test" }
    .flatMap("FlatMap") { line -> line.split(" ") }
    .mapFn("Word count", { word -> 1 / word.length })
    .result

Python

result = (CollectionComposer.of(input_teams)
          .map('Map with country', lambda name: TeamInfo(name=name, country=team_countries[name], city=''))
          .map('Map with city', lambda info: TeamInfo(name=info.name, country=info.country, city=team_cities[info.name]))
          .filter('Filter french teams', lambda info: info.country == 'France'))

What Asgarde does for you:

• Wraps the error handling logic of each step: try/catch blocks, tuple tags, exceptionsInto/exceptionsVia.
• Keeps the fluent style of Beam while collecting the failures of all the steps in a single PCollection.
• Gives access to the DoFn lifecycle (setup, start bundle, finish bundle, teardown) with simple actions.
• Handles errors in filters, which is not available with the Beam Filter transform in Java.
• Guarantees that the error handling itself can’t make the job fail, and counts the failures per step with Beam metrics.

And the Beam ErrorHandler?

The Beam Java SDK has an ErrorHandler: registered on the pipeline with a sink, it aggregates the bad records of several transforms into a single dead letter queue. The Beam IOs support it natively (e.g. KafkaIO.read().withBadRecordErrorHandler(handler), BigQueryIO.write().withErrorHandler(handler), PubsubIO.readStrings().withErrorHandler(handler)).

But the ErrorHandler is a place to put the errors: it doesn’t catch them in your transformation steps. The core transforms (MapElements, FlatMapElements, ParDo, Filter) don’t route their errors to an ErrorHandler, so each business step is a DoFn with its own try/catch block, BadRecordRouter, tagged outputs and coder.

The same two-step flow, with both:

// With the Beam ErrorHandler: each step is a DoFn catching its own errors.
class ParseFn extends DoFn<String, Order> {
    @ProcessElement
    public void process(@Element String json, MultiOutputReceiver out) throws Exception {
        try {
            out.get(ORDERS).output(OrderParser.parse(json));
        } catch (Exception e) {
            BadRecordRouter.RECORDING_ROUTER.route(out, json, StringUtf8Coder.of(), e, "Parse");
        }
    }
}
// ... and the same DoFn again for ValidateFn, with its own coder for the bad records.

final BadRecordErrorHandler<?> errorHandler = pipeline.registerBadRecordErrorHandler(deadLetterSink);

final PCollectionTuple parsed = input.apply("Parse", ParDo.of(new ParseFn())
        .withOutputTags(ORDERS, TupleTagList.of(BadRecordRouter.BAD_RECORD_TAG)));
errorHandler.addErrorCollection(parsed.get(BadRecordRouter.BAD_RECORD_TAG));

final PCollectionTuple validated = parsed.get(ORDERS).apply("Validate", ParDo.of(new ValidateFn())
        .withOutputTags(VALID_ORDERS, TupleTagList.of(BadRecordRouter.BAD_RECORD_TAG)));
errorHandler.addErrorCollection(validated.get(BadRecordRouter.BAD_RECORD_TAG));

errorHandler.close();

// With Asgarde: the same steps, the errors are caught for you.
final WithFailures.Result<PCollection<Order>, Failure> result = CollectionComposer.of(input)
        .apply("Parse", MapElementFn.into(TypeDescriptor.of(Order.class)).via(OrderParser::parse))
        .apply("Validate", FilterFn.by(Order::isValid))
        .getResult();

// And the failures can go to the same ErrorHandler, e.g. with the bad records of KafkaIO.
errorHandler.addErrorCollection(result.failures().apply("To bad records", FailureTransforms.toBadRecords()));

What each one brings:

	Beam ErrorHandler	Asgarde
Single dead letter queue for several steps	✅	✅
Bad records of the Beam IOs (Kafka, BigQuery, Pub/Sub…)	✅ native	✅ by adding the Asgarde failures to the ErrorHandler
Catching the errors of your transformation steps	❌ a DoFn with try/catch and BadRecordRouter per step	✅ built in, in the fluent apply chain
MapElements / FlatMapElements	❌ exceptionsInto / exceptionsVia and a conversion per step	✅ handled by the composer
Error handling in filters, side inputs, DoFn lifecycle actions	❌ to write in each DoFn	✅ FilterFn, MapProcessContextFn, lifecycle actions
Origin element (the element that entered the flow)	❌	✅ lazy, evaluated only on failure
Failure guarantees (non serializable exceptions, partial flatMap outputs, stable transform names)	❌	✅
Failure metrics per step	❌	✅
Kotlin extensions, Python version	❌	✅

In short: the ErrorHandler aggregates the errors and connects the IOs, Asgarde removes the error handling code of your transformation steps and brings its guarantees.

Asgarde and the ErrorHandler together

The two work together: the Beam IOs send their bad records to the ErrorHandler, the transformation steps are written with Asgarde, and the Asgarde failures are added to the same ErrorHandler. The whole pipeline gets a single dead letter queue, without error handling code in the business steps:

final BadRecordErrorHandler<?> errorHandler = pipeline.registerBadRecordErrorHandler(new WriteDeadLettersToBigQuery());

// 1. Read: the bad records of KafkaIO (e.g. deserialization errors) go to the ErrorHandler.
final PCollection<String> messages = pipeline
        .apply("Read orders", KafkaIO.<String, String>read()
                .withBootstrapServers(bootstrapServers)
                .withTopic("orders")
                .withKeyDeserializer(StringDeserializer.class)
                .withValueDeserializer(StringDeserializer.class)
                .withBadRecordErrorHandler(errorHandler)
                .withoutMetadata())
        .apply("Values", Values.create());

// 2. Transform: the business steps with Asgarde, the errors are caught for you.
final WithFailures.Result<PCollection<Order>, Failure> result = CollectionComposer.of(messages)
        .withOriginElement(message -> message)
        .apply("Parse", MapElementFn.into(TypeDescriptor.of(Order.class)).via(OrderParser::parse))
        .apply("Validate", FilterFn.by(Order::isValid))
        .getResult();

// ... and their failures go to the same ErrorHandler.
errorHandler.addErrorCollection(result.failures().apply("To bad records", FailureTransforms.toBadRecords()));

// 3. Write: the bad records of BigQueryIO go to the ErrorHandler too.
result.output()
        .apply("To rows", MapElements.into(TypeDescriptor.of(TableRow.class)).via(Order::toTableRow))
        .apply("Write orders", BigQueryIO.writeTableRows()
                .to("my-project:my_dataset.orders")
                .withErrorHandler(errorHandler));

// A single dead letter queue for the IOs and the business steps.
errorHandler.close();

The Asgarde failures keep their benefits in the BadRecords: the origin element as the record to replay, and the encoded elements with their coder. To keep the richer Failure objects (input and origin elements separately, per step metrics), write them to their own sink instead, see Write the failures and Beam native error handling.