[BUG] Error when building buffer for pipeline using Kafka

[bpalermo]

Describe the bug
A clear and concise description of what the bug is.

To Reproduce
Enable Kafka buffer for pipeline:

service-map-pipeline:
  ...
  buffer:
    kafka:
      bootstrap_servers: []
      topics:
      - name: topic-name
        group_id: kafka_group_id
        create_topic: true
      authentication:
        sasl:
          aws_msk_iam: default
      aws:
        region: aws_region
        msk:
          arn: msk_arn

Expected behavior
A clear and concise description of what you expected to happen.
Should not fail.

Screenshots
If applicable, add screenshots to help explain your problem.
N/A

Environment (please complete the following information):

• Version: 2.6.1

Additional context

2023-12-14T14:43:03,300 [main] INFO  org.opensearch.dataprepper.parser.PipelineTransformer - Building buffer for the pipeline [service-map-pipeline]
2023-12-14T14:43:03,302 [main] ERROR org.opensearch.dataprepper.parser.PipelineTransformer - Construction of pipeline components failed, skipping building of pipeline [service-map-pipeline] and its connected pipelines
org.opensearch.dataprepper.model.plugin.InvalidPluginDefinitionException: Unable to create an argument for required plugin parameter type: interface org.opensearch.dataprepper.model.codec.ByteDecoder
	at org.opensearch.dataprepper.plugin.ComponentPluginArgumentsContext.lambda$createBeanSupplier$10(ComponentPluginArgumentsContext.java:117) ~[data-prepper-core-2.6.1.jar:?]
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
	at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:992) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:575) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260) ~[?:?]
	at java.base/java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:616) ~[?:?]
	at java.base/java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:622) ~[?:?]
	at org.opensearch.dataprepper.plugin.ComponentPluginArgumentsContext.createArguments(ComponentPluginArgumentsContext.java:87) ~[data-prepper-core-2.6.1.jar:?]
	at org.opensearch.dataprepper.plugin.PluginCreator.newPluginInstance(PluginCreator.java:46) ~[data-prepper-core-2.6.1.jar:?]
	at org.opensearch.dataprepper.plugin.DefaultPluginFactory.loadPlugin(DefaultPluginFactory.java:75) ~[data-prepper-core-2.6.1.jar:?]
	at org.opensearch.dataprepper.parser.PipelineTransformer.buildPipelineFromConfiguration(PipelineTransformer.java:117) ~[data-prepper-core-2.6.1.jar:?]
	at org.opensearch.dataprepper.parser.PipelineTransformer.transformConfiguration(PipelineTransformer.java:97) ~[data-prepper-core-2.6.1.jar:?]
	at org.opensearch.dataprepper.DataPrepper.<init>(DataPrepper.java:67) ~[data-prepper-core-2.6.1.jar:2.6.1]
	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:?]
	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77) ~[?:?]
	at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:?]
	at java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499) ~[?:?]
	at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:480) ~[?:?]
	at org.springframework.beans.BeanUtils.instantiateClass(BeanUtils.java:211) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:117) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:311) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.ConstructorResolver.autowireConstructor(ConstructorResolver.java:296) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.autowireConstructor(AbstractAutowireCapableBeanFactory.java:1372) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1222) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:582) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:542) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:335) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:234) [spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:333) [spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:208) [spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:955) [spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:920) [spring-context-5.3.28.jar:5.3.28]
	at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:583) [spring-context-5.3.28.jar:5.3.28]
	at org.opensearch.dataprepper.AbstractContextManager.start(AbstractContextManager.java:59) [data-prepper-core-2.6.1.jar:2.6.1]
	at org.opensearch.dataprepper.AbstractContextManager.getDataPrepperBean(AbstractContextManager.java:45) [data-prepper-core-2.6.1.jar:2.6.1]
	at org.opensearch.dataprepper.DataPrepperExecute.main(DataPrepperExecute.java:39) [data-prepper-main-2.6.1.jar:2.6.1]
Caused by: org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'org.opensearch.dataprepper.model.codec.ByteDecoder' available
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBean(DefaultListableBeanFactory.java:351) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBean(DefaultListableBeanFactory.java:342) ~[spring-beans-5.3.28.jar:5.3.28]
	at org.opensearch.dataprepper.plugin.ComponentPluginArgumentsContext.lambda$createBeanSupplier$10(ComponentPluginArgumentsContext.java:115) ~[data-prepper-core-2.6.1.jar:?]
	... 38 more

[dlvenable]

@bpalermo , What source are you using for this pipeline?

[dlvenable]

@bpalermo , If this is a standard trace pipeline, we recommend that you use the kafka buffer on the entry-pipeline. Then use the in-memory buffer for the two connected pipelines (service-map-pipeline and raw-trace-pipeline).

[bpalermo]

@bpalermo , What source are you using for this pipeline?

entry-pipeline:
  ...
  source:
    otel_trace_source:

@bpalermo , If this is a standard trace pipeline, we recommend that you use the kafka buffer on the entry-pipeline. Then use the in-memory buffer for the two connected pipelines (service-map-pipeline and raw-trace-pipeline).

That's currently how it's being used, only on the entry pipeline and it's working.

[bpalermo]

@dlvenable any updates? I hit the same issue if I try to use kafka as source for my logs pipeline when reading from a topic written by the otlp exporter from Otel Collector.