Adding support for storing raw bytes in Kafka Buffer #3519

kkondaka · 2023-10-17T22:26:48Z

Description

Adding support for storing raw bytes in Kafka Buffer.

Added writeBytes api to Buffer
Added ByteDecoder that's set in the buffer based on the source during pipeline creation
HTTP source modified to call writeBytes API
Kafka buffer producer writes bytes to kafka
Kafka buffer consumer reads bytes and invokes decoder to create records

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

[X ] New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
[ X] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable · 2023-10-18T13:52:54Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/ByteDecoder.java

+     */
+    void parse(InputStream inputStream, Consumer<Record<Event>> eventConsumer) throws IOException;
+
+   // ByteDecoder getDecoder();


Let's remove this.

dlvenable · 2023-10-18T13:55:08Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/source/Source.java

@@ -35,4 +36,8 @@ public interface Source<T extends Record<?>> {
    default boolean areAcknowledgementsEnabled() {
        return false;
    }
+
+    default ByteDecoder getDecoder() {


I think we should make this a new interface instead of adding to Source. These could potentially be decoupled in the future if we saw the need.

public interface HasByteDecoder { ByteDecoder getDecoder(); }

dlvenable · 2023-10-18T13:58:10Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/PluginFactory.java

+     * @return A new instance of your plugin, configured
+     * @since 2.6
+     */
+    <T> T loadPlugin(final Class<T> baseClass, final PluginSetting pluginSetting, Object argument);


We can avoid a new interface and prepare for future extensions by using varargs here. There is no reason that we might not have more arguments in the future.

You can modify the existing loadPlugin(Class<T>, PluginSetting) by adding Object...

<T> T loadPlugin(final Class<T> baseClass, final PluginSetting pluginSetting, Object... arguments);

Existing calls will remain compatible because zero arguments are allowed.

dlvenable · 2023-10-18T13:59:43Z

...er-core/src/main/java/org/opensearch/dataprepper/plugin/ComponentPluginArgumentsContext.java

@@ -77,7 +81,13 @@ private ComponentPluginArgumentsContext(final Builder builder) {
    }

    @Override
-    public Object[] createArguments(final Class<?>[] parameterTypes) {
+    public Object[] createArguments(final Class<?>[] parameterTypes, Optional<Object> optionalArgument) {


Do not create Optional<> parameters. This is an anti-pattern. There is not guarantee that Optional is not null. So, you would need to write code like:

if(optionalSometing != null && optionalSomething.isPresent())

Also, make this a varargs as well. Object....

dlvenable · 2023-10-18T14:01:11Z

data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/PluginArgumentsContext.java

 interface PluginArgumentsContext {
-    Object[] createArguments(final Class<?>[] parameterTypes);
+    Object[] createArguments(final Class<?>[] parameterTypes, final Optional<Object> optionalArgument);


Do not create Optional<> parameters. This is an anti-pattern. There is not guarantee that Optional is not null. So, you would need to write code like:

if(optionalSometing != null && optionalSomething.isPresent())

Also, make this a varargs as well. Object....

dlvenable · 2023-10-18T14:12:44Z

.../http-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/HTTPSource.java


    @DataPrepperPluginConstructor
    public HTTPSource(final HTTPSourceConfig sourceConfig, final PluginMetrics pluginMetrics, final PluginFactory pluginFactory,
                      final PipelineDescription pipelineDescription) {
        this.sourceConfig = sourceConfig;
        this.pluginMetrics = pluginMetrics;
        this.pipelineName = pipelineDescription.getPipelineName();
+        this.byteDecoder = new JsonDecoder();


Right now, there is a slightly different decoder for the byte buffer versus the codec. This could lead to hard-to-find bugs.

We should use the exact same code path for decoding regardless of the buffer type. Please refactor this to connect the ByteDecoder directly to the codec.

dlvenable · 2023-10-18T14:15:31Z

...p-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/LogHTTPService.java

+            try {
+                buffer.writeAll(records, bufferWriteTimeoutInMillis);
+            } catch (Exception e) {
+                LOG.error("Failed to write the request of size {} due to: {}", content.length(), e.getMessage());


I think we should perform the writeBytes and writeAll in the same try-catch. This way, we have the same error handling on both. This split can lead to programming errors quite easily.

dlvenable · 2023-10-18T14:17:20Z

...p-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/LogHTTPService.java

+
+            try {
+                jsonList = jsonCodec.parse(content);
+            } catch (IOException e) {


I'm a bit concerned about not doing any parsing even with the byte buffer. With this change, callers will not get a Bad Request when sending bad data. It will only become apparent downstream. This means that a different buffer results in different fundamental behavior between pipelines. This doesn't seem quite right.

Do you suggest doing parse and ignore the parse result in case of byte buffer?

dlvenable · 2023-10-18T14:20:56Z

...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

-                            topicMetrics.getNumberOfBufferSizeOverflows().increment();
-                            Thread.sleep(100);
-                        }
+                if (schema == MessageFormat.BYTES && byteDecoder != null) {


Can we make use of the new SerializationFactory or add a similar concept to avoid having these conditionals in the consumer? Maybe a future improvement.

Yeah. we need to better organize it. I will open an issue for this.

dlvenable · 2023-10-18T14:21:28Z

...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/producer/KafkaCustomProducer.java

+        try {
+            send(topicName, key, bytes).get();
+        } catch (Exception e) {
+            LOG.error("Error occurred while publishing " + e.getMessage());


Use string interpolation with log lines:

LOG.error("Error occurred while publishing {}", e.getMessage());

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable

Overall looks good. I have a few comments, and I think we can also expand some test cases.

dlvenable · 2023-10-19T14:38:45Z

...er-core/src/main/java/org/opensearch/dataprepper/plugin/ComponentPluginArgumentsContext.java

-                ByteDecoder byteDecoder = (ByteDecoder) optionalArgument.get();
+    public Object[] createArguments(final Class<?>[] parameterTypes, final Object ... args) {
+        Map<Class<?>, Supplier<Object>> optionalArgumentsSuppliers = new HashMap<>();
+        Boolean byteDecoderInitialized = false;


The plugin framework shouldn't know or care about ByteDecoder at all. Any special code for that should be in PipelineParser.

Also, I'm not sure what this condition is really doing other than possibly avoiding passing the parameter in twice.

I think this code can be:

Map<Class<?>, Supplier<Object>> optionalArgumentsSuppliers = new HashMap<>(); for(final Object arg: args) { optionalArgumentsSuppliers.put(arg.getClass(), () -> arg); }

dlvenable · 2023-10-19T14:39:17Z

...er-core/src/main/java/org/opensearch/dataprepper/plugin/ComponentPluginArgumentsContext.java

                optionalArgumentsSuppliers.put(ByteDecoder.class, () -> byteDecoder);
            }
        }
        return Arrays.stream(parameterTypes)
-                .map(this::getRequiredArgumentSupplier)
+                .map((parameterType) -> getRequiredArgumentSupplier(parameterType, optionalArgumentsSuppliers))


nit: You don't need the parenthesis around parameterType.

.map(parameterType -> getRequiredArgumentSupplier(parameterType, optionalArgumentsSuppliers))

dlvenable

In addition to my previous comments, I've added some specific unit test cases we can cover.

Also, we should cover this change in DefaultPluginFactoryIT.

dlvenable · 2023-10-19T16:21:30Z

data-prepper-core/src/test/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryTest.java

@@ -256,7 +256,7 @@ void loadPlugins_should_return_a_single_instance_when_the_the_numberOfInstances_
            verify(pluginCreator).newPluginInstance(eq(expectedPluginClass), pluginArgumentsContextArgCapture.capture(), eq(pluginName));
            final ComponentPluginArgumentsContext actualPluginArgumentsContext = pluginArgumentsContextArgCapture.getValue();
            final List<Class> classes = List.of(PipelineDescription.class);
-            final Object[] pipelineDescriptionObj = actualPluginArgumentsContext.createArguments(classes.toArray(new Class[1]));
+            final Object[] pipelineDescriptionObj = actualPluginArgumentsContext.createArguments(classes.toArray(new Class[1]), Optional.empty());


Can you remove this Optional.empty()?

dlvenable · 2023-10-19T16:21:33Z

data-prepper-core/src/test/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryTest.java

@@ -292,7 +292,7 @@ void loadPlugins_should_return_an_instance_for_the_total_count() {
            assertThat(actualPluginArgumentsContextList.size(), equalTo(3));
            actualPluginArgumentsContextList.forEach(pluginArgumentsContext -> {
                final List<Class> classes = List.of(PipelineDescription.class);
-                final Object[] pipelineDescriptionObj = pluginArgumentsContext.createArguments(classes.toArray(new Class[1]));
+                final Object[] pipelineDescriptionObj = pluginArgumentsContext.createArguments(classes.toArray(new Class[1]), Optional.empty());


Can you remove this Optional.empty()?

dlvenable · 2023-10-19T16:21:52Z

data-prepper-core/src/test/java/org/opensearch/dataprepper/plugin/ExtensionLoaderTest.java

@@ -174,7 +175,7 @@ void loadExtensions_invokes_newPluginInstance_with_PluginArgumentsContext_not_su
        final PluginArgumentsContext actualPluginArgumentsContext = contextArgumentCaptor.getValue();

        final Class[] inputClasses = {String.class};
-        assertThrows(InvalidPluginDefinitionException.class, () -> actualPluginArgumentsContext.createArguments(inputClasses));
+        assertThrows(InvalidPluginDefinitionException.class, () -> actualPluginArgumentsContext.createArguments(inputClasses, Optional.empty()));


Can you remove this Optional.empty()?

Also, we might want a couple variations on the test.

createArguments(inputClasses)

and

final Object someArgument = UUID.randomUUID().toString(); createArguments(inputClasses, someArgument)

dlvenable · 2023-10-19T16:23:23Z

data-prepper-core/src/test/java/org/opensearch/dataprepper/plugin/ExtensionLoaderTest.java


        final PluginArgumentsContext actualPluginArgumentsContext = contextArgumentCaptor.getValue();

-        final Object[] arguments = actualPluginArgumentsContext.createArguments(new Class[]{});
+        final Object[] arguments = actualPluginArgumentsContext.createArguments(new Class[]{}, Optional.empty());


Remove the Optional.empty(). Also, we might want two test cases here.

dlvenable · 2023-10-19T16:24:05Z

data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/DefaultPluginFactory.java

        final String pluginName = pluginSetting.getName();
        final Class<? extends T> pluginClass = getPluginClass(baseClass, pluginName);

        final ComponentPluginArgumentsContext constructionContext = getConstructionContext(pluginSetting, pluginClass, null);

-        return pluginCreator.newPluginInstance(pluginClass, constructionContext, pluginName);
+        return pluginCreator.newPluginInstance(pluginClass, constructionContext, pluginName, args);


Let's have some test cases to verify that this works with args and without args. And that these args are passed into the correct mocks.

dlvenable · 2023-10-19T16:24:41Z

...er-core/src/main/java/org/opensearch/dataprepper/plugin/ComponentPluginArgumentsContext.java

@@ -77,21 +78,30 @@ private ComponentPluginArgumentsContext(final Builder builder) {
    }

    @Override
-    public Object[] createArguments(final Class<?>[] parameterTypes) {
+    public Object[] createArguments(final Class<?>[] parameterTypes, final Object ... args) {


Also, let's add unit tests for this method to ensure that args are used correctly and as expected.

Signed-off-by: Krishna Kondaka <[email protected]>

graytaylor0 · 2023-10-20T22:54:33Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/buffer/AbstractBuffer.java

@@ -112,6 +112,11 @@ public void writeAll(Collection<T> records, int timeoutInMillis) throws Exceptio
        }
    }

+    @Override
+    public void writeBytes(final byte[] bytes, final String key, int timeoutInMillis) throws Exception {
+        throw new RuntimeException("not supported");


There is an UnsupportedOperationException

Will do this in the next PR

Adding support for storing raw bytes in Kafka Buffer

315e6b2

Signed-off-by: Krishna Kondaka <[email protected]>

kkondaka requested review from chenqi0805, engechas, graytaylor0, dinujoh, asifsmohammed, dlvenable and oeyh as code owners October 17, 2023 22:26

Modified to wait for the send() to finish before returning

ad8faa3

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable requested changes Oct 18, 2023

View reviewed changes

Krishna Kondaka added 4 commits October 18, 2023 22:36

Addressed review comments

fb82aca

Signed-off-by: Krishna Kondaka <[email protected]>

Removed unused imports

4719a83

Signed-off-by: Krishna Kondaka <[email protected]>

Fixed Kafka integration test

5b562c3

Signed-off-by: Krishna Kondaka <[email protected]>

Fixed json processor check style errors

5a98979

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable requested changes Oct 19, 2023

View reviewed changes

Krishna Kondaka added 2 commits October 19, 2023 19:47

Addressed review comments and added a new test case

dd535dd

Signed-off-by: Krishna Kondaka <[email protected]>

Addressed review comments and added a new tests

483c922

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable approved these changes Oct 20, 2023

View reviewed changes

graytaylor0 approved these changes Oct 20, 2023

View reviewed changes

kkondaka merged commit aa81607 into opensearch-project:main Oct 20, 2023
26 of 27 checks passed

kkondaka deleted the kafka-bytes branch May 13, 2024 05:52

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding support for storing raw bytes in Kafka Buffer #3519

Adding support for storing raw bytes in Kafka Buffer #3519

kkondaka commented Oct 17, 2023

dlvenable Oct 18, 2023

dlvenable Oct 18, 2023

dlvenable Oct 18, 2023

dlvenable Oct 18, 2023

dlvenable Oct 18, 2023

dlvenable Oct 18, 2023

dlvenable Oct 18, 2023

dlvenable Oct 18, 2023

kkondaka Oct 18, 2023

dlvenable Oct 18, 2023

kkondaka Oct 18, 2023

dlvenable Oct 18, 2023

dlvenable left a comment

dlvenable Oct 19, 2023

dlvenable Oct 19, 2023

dlvenable left a comment

dlvenable Oct 19, 2023

dlvenable Oct 19, 2023

dlvenable Oct 19, 2023

dlvenable Oct 19, 2023

dlvenable Oct 19, 2023

dlvenable Oct 19, 2023

dlvenable Oct 19, 2023

graytaylor0 Oct 20, 2023

kkondaka Oct 20, 2023

Adding support for storing raw bytes in Kafka Buffer #3519

Adding support for storing raw bytes in Kafka Buffer #3519

Conversation

kkondaka commented Oct 17, 2023

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment