From c673b66948a7904abf403810bd61dfd3e15daff6 Mon Sep 17 00:00:00 2001 From: Jonah Calvo Date: Wed, 6 Aug 2025 12:07:13 -0500 Subject: [PATCH] Convert null message to empty string in DLQ to avoid NPE Signed-off-by: Jonah Calvo --- .../sink/opensearch/OpenSearchSink.java | 2 +- .../sink/opensearch/OpenSearchSinkTest.java | 44 +++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java index 4fa8eb5926..5168861a2b 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java +++ b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java @@ -679,7 +679,7 @@ private DlqObject createDlqObjectFromEvent(final Event event, .withFailedData(FailedDlqData.builder() .withDocument(event.toJsonString()) .withIndex(index) - .withMessage(message) + .withMessage(message != null ? message : "") .build()) .withPluginName(PLUGIN_NAME) .withPipelineName(pipeline) diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkTest.java index 59019e0b0a..191951f0e1 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkTest.java +++ b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkTest.java @@ -346,4 +346,48 @@ void doOutput_with_invalid_version_expression_result_catches_RuntimeException_an verify(dynamicDocumentVersionDroppedEvents).increment(); } + + @Test + void createDlqObjectFromEvent_with_null_message_uses_default_message() throws IOException { + when(pluginSetting.getName()).thenReturn("opensearch"); + + final Event event = mock(JacksonEvent.class); + final String document = UUID.randomUUID().toString(); + when(event.toJsonString()).thenReturn(document); + final EventHandle eventHandle = mock(EventHandle.class); + when(event.getEventHandle()).thenReturn(eventHandle); + final String index = UUID.randomUUID().toString(); + + final OpenSearchSink objectUnderTest = createObjectUnderTest(); + when(indexManagerFactory.getIndexManager(any(IndexType.class), eq(openSearchClient), any(RestHighLevelClient.class), eq(openSearchSinkConfiguration), any(TemplateStrategy.class), any())) + .thenReturn(indexManager); + doNothing().when(indexManager).setupIndex(); + objectUnderTest.initialize(); + + final DlqObject.Builder dlqObjectBuilder = mock(DlqObject.Builder.class); + final ArgumentCaptor failedDlqData = ArgumentCaptor.forClass(FailedDlqData.class); + when(dlqObjectBuilder.withEventHandle(eventHandle)).thenReturn(dlqObjectBuilder); + when(dlqObjectBuilder.withFailedData(failedDlqData.capture())).thenReturn(dlqObjectBuilder); + when(dlqObjectBuilder.withPluginName(pluginSetting.getName())).thenReturn(dlqObjectBuilder); + when(dlqObjectBuilder.withPluginId(pluginSetting.getName())).thenReturn(dlqObjectBuilder); + when(dlqObjectBuilder.withPipelineName(pipelineDescription.getPipelineName())).thenReturn(dlqObjectBuilder); + when(dlqObjectBuilder.build()).thenReturn(mock(DlqObject.class)); + + try (final MockedStatic dlqObjectMockedStatic = mockStatic(DlqObject.class)) { + dlqObjectMockedStatic.when(DlqObject::builder).thenReturn(dlqObjectBuilder); + + // Use reflection to call the private method with null message + java.lang.reflect.Method method = OpenSearchSink.class.getDeclaredMethod("createDlqObjectFromEvent", Event.class, String.class, String.class); + method.setAccessible(true); + method.invoke(objectUnderTest, event, index, null); + } catch (Exception e) { + throw new RuntimeException(e); + } + + final FailedDlqData failedDlqDataResult = failedDlqData.getValue(); + assertThat(failedDlqDataResult, notNullValue()); + assertThat(failedDlqDataResult.getDocument(), equalTo(document)); + assertThat(failedDlqDataResult.getIndex(), equalTo(index)); + assertThat(failedDlqDataResult.getMessage(), equalTo("")); + } }