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();