I have an apache beam pipeline that accepts messages then saves the message to a queue or database. If saving to Cassandra fails due to database failure or other issues how would I save the original message to a different queue? Here is how I am creating the pipeline

public Pipeline pipelineDirectRunner() {
        PipelineOptions options = PipelineOptionsFactory.create();
        options.setRunner(DirectRunner.class);
        return  Pipeline.create(options);
}

0 Answers