ERROR org.opensearch.dataprepper.GrpcRequestExceptionHandler in dataprepper version 2.7.0

Describe the bug
We have an Data Prepper & OpenTelemetry setup in our Kubernetes ecosystem running to collect metrics and traces and send the data to OpenSearch. This setup was working perfectly fine when we were running with below versions.

  Data Prepper - 2.6.0
  OpenTelemetry Collector - 0.83.0

Recently we had performed a version upgrade of dataprepper component only to remove certain vulnerabilities associated with image. The new versions we are on are as below

  Data Prepper -  2.7.0
  OpenTelemetry Collector - 0.83.0

However post this upgrade we are encountering below error in Data Prepper and metrics / traces are not reaching the Opensearch.

 [armeria-common-worker-epoll-3-2] ERROR org.opensearch.dataprepper.GrpcRequestExceptionHandler - Unexpected 
 exception handling gRPC request
 com.linecorp.armeria.common.stream.ClosedStreamException: received a RST_STREAM frame: CANCEL
 [pool-9-thread-94] ERROR org.opensearch.dataprepper.plugins.source.otelmetrics.OTelMetricsGrpcService - Failed to 
 write the request of size 120068 due to:
 java.util.concurrent.TimeoutException: Pipeline [otel-metrics-pipeline] - Buffer does not have enough capacity left for the 
 number of records: 286, timed out waiting for slots.
at org.opensearch.dataprepper.plugins.buffer.blockingbuffer.BlockingBuffer.doWriteAll(BlockingBuffer.java:127) ~ 
 [blocking-buffer-2.7.0.jar:?]
at org.opensearch.dataprepper.model.buffer.AbstractBuffer.writeAll(AbstractBuffer.java:107) ~[data-prepper-api-2.7.0.jar:?]
at org.opensearch.dataprepper.model.buffer.DelegatingBuffer.writeAll(DelegatingBuffer.java:48) ~[data-prepper-api-2.7.0.jar:?]
at org.opensearch.dataprepper.model.buffer.DelegatingBuffer.writeAll(DelegatingBuffer.java:48) ~[data-prepper-api-2.7.0.jar:?]
at org.opensearch.dataprepper.parser.CircuitBreakingBuffer.writeAll(CircuitBreakingBuffer.java:50) ~[data-prepper-core-2.7.0.jar:?]
at org.opensearch.dataprepper.plugins.source.otelmetrics.OTelMetricsGrpcService.processRequest(OTelMetricsGrpcService.java:97) ~[otel-metrics-source-2.7.0.jar:?]
at org.opensearch.dataprepper.plugins.source.otelmetrics.OTelMetricsGrpcService.lambda$export$0(OTelMetricsGrpcService.java:83) ~[otel-metrics-source-2.7.0.jar:?]
at io.micrometer.core.instrument.composite.CompositeTimer.record(CompositeTimer.java:141) ~[micrometer-core-1.11.5.jar:1.11.5]
at org.opensearch.dataprepper.plugins.source.otelmetrics.OTelMetricsGrpcService.export(OTelMetricsGrpcService.java:83) ~[otel-metrics-source-2.7.0.jar:?]
at io.opentelemetry.proto.collector.metrics.v1.MetricsServiceGrpc$MethodHandlers.invoke(MetricsServiceGrpc.java:246) ~[opentelemetry-proto-0.16.0-alpha.jar:0.16.0]
at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:182) ~[grpc-stub-1.58.0.jar:1.58.0]
at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.invokeOnMessage(AbstractServerCall.java:387) ~[armeria-grpc-1.26.4.jar:?]
at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.lambda$onRequestMessage$2(AbstractServerCall.java:351) ~[armeria-grpc-1.26.4.jar:?]
at com.linecorp.armeria.internal.shaded.guava.util.concurrent.SequentialExecutor$1.run(SequentialExecutor.java:125) [armeria-1.26.4.jar:?]
at com.linecorp.armeria.internal.shaded.guava.util.concurrent.SequentialExecutor$QueueWorker.workOnQueue(SequentialExecutor.java:237) [armeria-1.26.4.jar:?]
at com.linecorp.armeria.internal.shaded.guava.util.concurrent.SequentialExecutor$QueueWorker.run(SequentialExecutor.java:182) [armeria-1.26.4.jar:?]
at com.linecorp.armeria.common.DefaultContextAwareRunnable.run(DefaultContextAwareRunnable.java:45) [armeria-1.26.4.jar:?]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.base/java.lang.Thread.run(Thread.java:840) [?:?] 
   [armeria-common-worker-epoll-3-1] WARN  io.netty.util.concurrent.AbstractEventExecutor - A task raised an exception. 
   Task: com.linecorp.armeria.common.DefaultContextAwareRunnable@23d5df32
   java.lang.IllegalStateException: call already closed. status: Status{code=RESOURCE_EXHAUSTED, description=Pipeline 
  [otel-metrics-pipeline] - Buffer does not have enough capacity left for the number of records: 286, timed out waiting for 
  slots., cause=null}, exception: org.opensearch.dataprepper.exceptions.BufferWriteException: Pipeline [otel-metrics- 
  pipeline] - Buffer does not have enough capacity left for the number of records: 286, timed out waiting for slots.
at com.linecorp.armeria.internal.shaded.guava.base.Preconditions.checkState(Preconditions.java:835) ~[armeria-1.26.4.jar:?]
at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.doClose(AbstractServerCall.java:245) ~[armeria-grpc-1.26.4.jar:?]
at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.lambda$close$1(AbstractServerCall.java:227) ~[armeria-grpc-1.26.4.jar:?]
at com.linecorp.armeria.common.DefaultContextAwareRunnable.run(DefaultContextAwareRunnable.java:45) ~[armeria-1.26.4.jar:?]
at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173) ~[netty-common-4.1.100.Final.jar:4.1.100.Final]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166) [netty-common-4.1.100.Final.jar:4.1.100.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) [netty-common-4.1.100.Final.jar:4.1.100.Final]
at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:413) [netty-transport-classes-epoll-4.1.100.Final.jar:4.1.100.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) [netty-common-4.1.100.Final.jar:4.1.100.Final]
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [netty-common-4.1.100.Final.jar:4.1.100.Final]
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.100.Final.jar:4.1.100.Final]
at java.base/java.lang.Thread.run(Thread.java:840) [?:?]

To Reproduce

Bring up similar setup on kubernetes ecosystem to collect and send metrics & traces to Opensearch.

  Data Prepper -  2.7.0
  OpenTelemetry Collector - 0.83.0

Expected behavior

Metrics and traces collected from kubernetes need to flow to opensearch seamleslly,

Start by reproducing the Kubernetes setup with Data Prepper 2.7.0 and OpenTelemetry Collector 0.83.0, then inspect BlockingBuffer.java, CircuitBreakingBuffer.java, and OTelMetricsGrpcService.java using the reported stack trace. Done means identifying why the buffer times out and verifying that collected metrics and traces reach OpenSearch without the reported gRPC errors.