From 0a6a548340b518669249a6865c7e899ef75915c3 Mon Sep 17 00:00:00 2001 From: Taylor Gray Date: Wed, 7 Oct 2026 10:29:34 -0500 Subject: [PATCH] fix(core): Skip circuit breaker for DLQ pipeline buffer The DLQ (failure) pipeline uses a HeadlessPipelineSource which writes events to its buffer via sendEvents(). When a circuit breaker is active, the buffer is wrapped with CircuitBreakingBuffer which throws TimeoutException when the breaker is open. This causes DLQ writes to block indefinitely (with acks enabled) or release with false (without acks), preventing acknowledgements from completing upstream. Since the DLQ pipeline exists specifically to drain failed events, blocking it under memory pressure defeats its purpose and can cascade into negative acknowledgements on the source pipeline's Kafka buffer. Skip applying CircuitBreakingBuffer to HeadlessPipelineSource pipelines, matching the existing skip for PipelineConnector sources. Signed-off-by: Taylor Gray --- .../core/parser/PipelineTransformer.java | 3 +++ .../core/parser/PipelineTransformerTests.java | 19 +++++++++++++++++++ 2 files changed, 22 insertions(+) diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java index aae8873646..8746f14ed9 100644 --- a/data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java +++ b/data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java @@ -449,6 +449,9 @@ private Buffer applyCircuitBreakerToBuffer(final Source source, final Buffer buf if (source instanceof PipelineConnector) return buffer; + if (source instanceof HeadlessPipelineSource) + return buffer; + if (buffer.isWrittenOffHeapOnly()) return buffer; diff --git a/data-prepper-core/src/test/java/org/opensearch/dataprepper/core/parser/PipelineTransformerTests.java b/data-prepper-core/src/test/java/org/opensearch/dataprepper/core/parser/PipelineTransformerTests.java index 1a5aeef838..fd27df0ca2 100644 --- a/data-prepper-core/src/test/java/org/opensearch/dataprepper/core/parser/PipelineTransformerTests.java +++ b/data-prepper-core/src/test/java/org/opensearch/dataprepper/core/parser/PipelineTransformerTests.java @@ -400,6 +400,25 @@ void testMultipleSinksWithFailurePipeline() { assertThat(((HeadlessPipelineSource)failurePipeline.getSource()).getAcknowledgementsEnabled(), equalTo(false)); } + @Test + void parseConfiguration_does_not_apply_circuit_breaker_to_failure_pipeline_buffer() { + final CircuitBreaker circuitBreaker = mock(CircuitBreaker.class); + when(circuitBreakerManager.getGlobalCircuitBreaker()) + .thenReturn(Optional.of(circuitBreaker)); + when(dataPrepperConfiguration.getFailurePipelineName()).thenReturn(DataPrepperConfiguration.DEFAULT_FAILURE_PIPELINE_NAME); + when(expressionEvaluator.isValidExpressionStatement("/value == raw")).thenReturn(true); + when(expressionEvaluator.isValidExpressionStatement("/value == service")).thenReturn(true); + mockDataPrepperConfigurationAccesses(); + final PipelineTransformer pipelineTransformer = + createObjectUnderTest(TestDataProvider.VALID_MULTIPLE_SINKS_WITH_FAILURE_PIPELINE_CONFIG_FILE); + final Map pipelineMap = pipelineTransformer.transformConfiguration(this.pipelinesDataFlowModel); + + final Pipeline failurePipeline = pipelineMap.get(DataPrepperConfiguration.DEFAULT_FAILURE_PIPELINE_NAME); + assertThat(failurePipeline, notNullValue()); + assertThat(failurePipeline.getSource(), instanceOf(HeadlessPipelineSource.class)); + assertThat(failurePipeline.getBuffer(), CoreMatchers.not(instanceOf(CircuitBreakingBuffer.class))); + } + @Test void testMultipleProcessors() { mockDataPrepperConfigurationAccesses();