Fix bugs that caused test to fail

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Author: jeffreyAaron

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/KafkaBuffer.java

@@ -83,7 +83,7 @@ public KafkaBuffer(final PluginSetting pluginSetting, final KafkaBufferConfig ka
         final String metricPrefixName = kafkaBufferConfig.getCustomMetricPrefix().orElse(pluginSetting.getName());
         final PluginMetrics producerMetrics = PluginMetrics.fromNames(metricPrefixName + WRITE, pluginSetting.getPipelineName());
         producer = kafkaCustomProducerFactory.createProducer(kafkaBufferConfig, null, null, producerMetrics, null, false);
-        final KafkaCustomConsumerFactory kafkaCustomConsumerFactory = new KafkaCustomConsumerFactory(serializationFactory, awsCredentialsSupplier);
+        final KafkaCustomConsumerFactory kafkaCustomConsumerFactory = new KafkaCustomConsumerFactory(serializationFactory, awsCredentialsSupplier, kafkaBufferConfig.getCompressionEnabled());
         innerBuffer = new BlockingBuffer<>(INNER_BUFFER_CAPACITY, INNER_BUFFER_BATCH_SIZE, pluginSetting.getPipelineName());
         this.shutdownInProgress = new AtomicBoolean(false);
         final PluginMetrics consumerMetrics = PluginMetrics.fromNames(metricPrefixName + READ, pluginSetting.getPipelineName());

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

@@ -9,6 +9,7 @@
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.luben.zstd.Zstd;
 import com.github.luben.zstd.ZstdBufferDecompressingStream;
 import com.github.luben.zstd.ZstdInputStream;
 import com.google.common.annotations.VisibleForTesting;
@@ -562,6 +563,7 @@ private <T> void iterateRecordPartitions(ConsumerRecords<String, T> records, fin
                     InputStream byteInputStream = new ByteArrayInputStream((byte[])consumerRecord.value());
                     InputStream inputStream;
                     if (compressionEnabled) {
+                        inputStream = new ZstdInputStream(byteInputStream);
                     } else {
                         inputStream = byteInputStream;
                     }

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/producer/KafkaCustomProducerFactory.java

@@ -96,16 +96,7 @@ public KafkaCustomProducer createProducer(final KafkaProducerConfig kafkaProduce
         final SchemaService schemaService = new SchemaService.SchemaServiceBuilder().getFetchSchemaService(topicName, kafkaProducerConfig.getSchemaConfig()).build();
         return new KafkaCustomProducer(producer,
             kafkaProducerConfig, dlqSink,
-            expressionEvaluator, Objects.nonNull(sinkContext) ? sinkContext.getTagsTargetKey() : null, topicMetrics, schemaService);
-    }
-
-    public KafkaCustomProducer createCompressedProducer(final KafkaProducerConfig kafkaProducerConfig,
-                                                        final ExpressionEvaluator expressionEvaluator, final SinkContext sinkContext, final PluginMetrics pluginMetrics,
-                                                        final DLQSink dlqSink,
-                                                        final boolean topicNameInMetrics) {
-        KafkaCustomProducer kafkaCustomProducer = createProducer(kafkaProducerConfig, expressionEvaluator, sinkContext, pluginMetrics, dlqSink, topicNameInMetrics);
-        kafkaCustomProducer.setCompressionEnabled(compressionEnabled);
-        return kafkaCustomProducer;
+            expressionEvaluator, Objects.nonNull(sinkContext) ? sinkContext.getTagsTargetKey() : null, topicMetrics, schemaService, compressionEnabled);
     }
 
     private void prepareTopicAndSchema(final KafkaProducerConfig kafkaProducerConfig, final Integer maxRequestSize) {