Revert HTTP data chunking changes for kafka buffer done in PR 4266 (#4329) (#4334)

Signed-off-by: Krishna Kondaka <[email protected]>
Co-authored-by: Krishna Kondaka <[email protected]>
(cherry picked from commit c8e010f)

Co-authored-by: Krishna Kondaka <[email protected]>

Author: opensearch-trigger-bot[bot]

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

@@ -20,7 +20,7 @@
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.source.Source;
 import org.opensearch.dataprepper.model.codec.ByteDecoder;
-import org.opensearch.dataprepper.model.codec.JsonObjectDecoder;
+import org.opensearch.dataprepper.model.codec.JsonDecoder;
 import com.linecorp.armeria.server.HttpService;
 import com.linecorp.armeria.server.Server;
 import com.linecorp.armeria.server.ServerBuilder;
@@ -64,7 +64,7 @@ public HTTPSource(final HTTPSourceConfig sourceConfig, final PluginMetrics plugi
         this.sourceConfig = sourceConfig;
         this.pluginMetrics = pluginMetrics;
         this.pipelineName = pipelineDescription.getPipelineName();
-        this.byteDecoder = new JsonObjectDecoder();
+        this.byteDecoder = new JsonDecoder();
         this.certificateProviderFactory = new CertificateProviderFactory(sourceConfig);
         final PluginModel authenticationConfiguration = sourceConfig.getAuthentication();
         final PluginSetting authenticationPluginSetting;

data-prepper-plugins/http-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/LogHTTPService.java

@@ -87,9 +87,9 @@ private HttpResponse processRequest(final AggregatedHttpRequest aggregatedHttpRe
         }
         try {
             if (buffer.isByteBuffer()) {
-                for (final String json: jsonList) {
-                    buffer.writeBytes(json.getBytes(), null, bufferWriteTimeoutInMillis);
-                }
+                // jsonList is ignored in this path but parse() was done to make 
+                // sure that the data is in the expected json format
+                buffer.writeBytes(content.array(), null, bufferWriteTimeoutInMillis);
             } else {
                 final List<Record<Log>> records = jsonList.stream()
                         .map(this::buildRecordLog)