Skip to content

Remove default codec and require codec for sqs sink #6486

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
graytaylor0 merged 2 commits into from
Mar 18, 2026
Merged

Remove default codec and require codec for sqs sink #6486

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4728dd4
Remove default codec and require codec for sqs sink from ndjson to json
graytaylor0 Feb 4, 2026
168c9b4
Remove validation for sqs sink codec with max_events_per_message
graytaylor0 Mar 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 5 additions & 10 deletions ...r-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Map;
import java.util.Collection;

@Experimental
Expand All @@ -60,16 +59,12 @@ public SqsSink(final PluginSetting pluginSetting,
sinkInitialized = false;
final PluginModel codecConfiguration = sqsSinkConfig.getCodec();
final PluginSetting codecPluginSettings;
if (codecConfiguration != null) {
String codecPluginName = codecConfiguration.getPluginName();
if (!codecPluginName.equals("json") && !codecPluginName.equals("ndjson")) {
throw new RuntimeException(String.format("Codec {} not supported.", codecPluginName));
}
codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(),
codecConfiguration.getPluginSettings());
} else {
codecPluginSettings = new PluginSetting("ndjson", Map.of());
String codecPluginName = codecConfiguration.getPluginName();
if (!codecPluginName.equals("json") && !codecPluginName.equals("ndjson")) {
Copy link
Copy Markdown
Member

@dlvenable dlvenable Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should allow for other codecs. I think the only time we want to be restrictive on the codec is when the customer is not using batching (ie. max_events_per_batch == 1). In that case, ndjson should be the only permissible codec.

Customers may want to serialize in Ion, XML, or other codecs.

Copy link
Copy Markdown
Member Author

@graytaylor0 graytaylor0 Feb 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is not clear why the codecs are so limited here

throw new RuntimeException(String.format("Codec {} not supported.", codecPluginName));
}
codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(),
codecConfiguration.getPluginSettings());

AwsConfig awsConfig = sqsSinkConfig.getAwsConfig();
final AwsCredentialsProvider awsCredentialsProvider = (awsConfig != null) ? awsCredentialsSupplier.getProvider(convertToCredentialOptions(awsConfig)) : awsCredentialsSupplier.getProvider(AwsCredentialsOptions.builder().build());
Expand Down
8 changes: 1 addition & 7 deletions ...ins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ public class SqsSinkConfig {
private String queueUrl;

@JsonProperty("codec")
@NotNull
private PluginModel codec;

@JsonProperty("threshold")
Expand Down Expand Up @@ -67,12 +68,5 @@ boolean isValidConfig() {
return (groupId == null && deDupId == null);
}
}

@AssertTrue(message = "ndjson codec (default codec) doesn't support max events per message greater than 1")
boolean isValidCodecConfig() {
if ((codec == null || codec.getPluginName().equals("ndjson")) && thresholdConfig.getMaxEventsPerMessage() > 1)
return false;
return true;
}
}

17 changes: 1 addition & 16 deletions ...sqs-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ void TestDefaultConfig() {
}

@Test
private void TestCustomConfig() throws Exception {
void TestCustomConfig() throws Exception {
AwsConfig awsConfig = mock(AwsConfig.class);
reflectivelySetField(sqsSinkConfig, "awsConfig", awsConfig);
assertThat(sqsSinkConfig.getAwsConfig(), equalTo(awsConfig));
Expand Down Expand Up @@ -114,21 +114,6 @@ void TestValidFiFoQConfigs() throws Exception {

}

@Test
void TestValidCodecConfig() throws Exception {
reflectivelySetField(sqsSinkConfig, "codec", null);
reflectivelySetField(sqsSinkConfig, "thresholdConfig", sqsThresholdConfig);
when(sqsThresholdConfig.getMaxEventsPerMessage()).thenReturn(2);
assertFalse(sqsSinkConfig.isValidCodecConfig());
when(sqsThresholdConfig.getMaxEventsPerMessage()).thenReturn(1);
assertTrue(sqsSinkConfig.isValidCodecConfig());
PluginModel codec = mock(PluginModel.class);
when(codec.getPluginName()).thenReturn("ndjson");
reflectivelySetField(sqsSinkConfig, "codec", codec);
when(sqsThresholdConfig.getMaxEventsPerMessage()).thenReturn(2);
assertFalse(sqsSinkConfig.isValidCodecConfig());
}

private void reflectivelySetField(final SqsSinkConfig sqsSinkConfig, final String fieldName, final Object value) throws NoSuchFieldException, IllegalAccessException {
final Field field = SqsSinkConfig.class.getDeclaredField(fieldName);
try {
Expand Down
14 changes: 0 additions & 14 deletions ...ugins/sqs-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,20 +230,6 @@ void TestForDefaultCodec() {
}
}

@Test
void TestForNullCodec() {
when(sqsSinkConfig.getCodec()).thenReturn(null);
try(MockedStatic<SqsClientFactory> mockedStatic = mockStatic(SqsClientFactory.class)) {
mockedStatic.when(() -> SqsClientFactory.createSqsClient(any(Region.class),
any(AwsCredentialsProvider.class)))
.thenReturn(sqsClient);

SqsSink sqsSink = createObjectUnderTest();
sqsSink.doInitialize();
assertTrue(sqsSink.isReady());
}
}

@Test
void TestSinkOutputWithEvents() {
try(MockedStatic<SqsClientFactory> mockedStatic = mockStatic(SqsClientFactory.class)) {
Expand Down
Loading