[BUG] Java process does not get killed when data prepper itself terminates

Describe the bug
The running java process does not get killed when data prepper itself terminates. This is problematic when running data prepper inside of a docker container using kubernetes HorizontalPodAutoscaler to control data prepper. Right now data prepper will terminate for some reason however the container/pod does not terminate and will stick around as a zombie pod- holding onto resources but unable to function. I believe this is because PID 1 (the java cmd running data prepper) does not get terminated when the data-prepper process stops.

To Reproduce
Really you just need data prepper to try to use more resources than is available in a container/pod. Here is one way

1. Run data prepper in kubernetes
2. Set the pod resource limit to something low like cpu 100m to force it to run into resource issues
3. Configure data prepper to accept data
4. Fire a lot of data at data prepper really fast (One way is to build up a buffer in your source and then try to send it to data prepper all at once) What I did was force restart data prepper a few times while simultaneously generated a bunch of source data trying to send to data prepper. Then data prepper gets overwhelmed, tries to get more resources, throws an error and terminates

Sample error

2023-04-03T17:18:30,799 [pool-7-thread-83] ERROR com.amazon.dataprepper.plugins.source.otellogs.OTelLogsGrpcService - Failed to write the request of size 2380052 due to:
java.util.concurrent.TimeoutException: Pipeline [log-pipeline] - Buffer does not have enough capacity left for the size of records: 2611, timed out waiting for slots.
	at org.opensearch.dataprepper.plugins.buffer.blockingbuffer.BlockingBuffer.doWriteAll(BlockingBuffer.java:123) ~[blocking-buffer-2.1.0.jar:?]
	at org.opensearch.dataprepper.model.buffer.AbstractBuffer.writeAll(AbstractBuffer.java:100) ~[data-prepper-api-2.1.0.jar:?]
	at org.opensearch.dataprepper.plugins.MultiBufferDecorator.writeAll(MultiBufferDecorator.java:39) ~[data-prepper-core-2.1.0.jar:?]
	at com.amazon.dataprepper.plugins.source.otellogs.OTelLogsGrpcService.processRequest(OTelLogsGrpcService.java:106) ~[otel-logs-source-2.1.0.jar:?]
	at com.amazon.dataprepper.plugins.source.otellogs.OTelLogsGrpcService.lambda$export$0(OTelLogsGrpcService.java:80) ~[otel-logs-source-2.1.0.jar:?]
	at io.micrometer.core.instrument.composite.CompositeTimer.record(CompositeTimer.java:141) ~[micrometer-core-1.10.3.jar:1.10.3]
	at com.amazon.dataprepper.plugins.source.otellogs.OTelLogsGrpcService.export(OTelLogsGrpcService.java:80) ~[otel-logs-source-2.1.0.jar:?]
	at io.opentelemetry.proto.collector.logs.v1.LogsServiceGrpc$MethodHandlers.invoke(LogsServiceGrpc.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.49.1.jar:1.49.1]
	at com.linecorp.armeria.server.grpc.AbstractServerCall.invokeOnMessage(AbstractServerCall.java:384) ~[armeria-grpc-1.22.1.jar:?]
	at com.linecorp.armeria.server.grpc.AbstractServerCall.lambda$onRequestMessage$2(AbstractServerCall.java:348) ~[armeria-grpc-1.22.1.jar:?]
	at com.linecorp.armeria.internal.shaded.guava.util.concurrent.SequentialExecutor$1.run(SequentialExecutor.java:123) ~[armeria-1.22.1.jar:?]
	at com.linecorp.armeria.internal.shaded.guava.util.concurrent.SequentialExecutor$QueueWorker.workOnQueue(SequentialExecutor.java:235) ~[armeria-1.22.1.jar:?]
	at com.linecorp.armeria.internal.shaded.guava.util.concurrent.SequentialExecutor$QueueWorker.run(SequentialExecutor.java:180) ~[armeria-1.22.1.jar:?]
	at com.linecorp.armeria.common.RequestContext.lambda$makeContextAware$3(RequestContext.java:566) ~[armeria-1.22.1.jar:?]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
	at java.lang.Thread.run(Thread.java:833) ~[?:?]
2023-04-03T17:18:30,800 [log-pipeline-sink-worker-4-thread-1] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [log-pipeline] - Shutting down process workers
2023-04-03T17:19:01,806 [log-pipeline-sink-worker-4-thread-1] WARN  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [log-pipeline] - Workers did not terminate in time, forcing termination
2023-04-03T17:19:05,278 [log-pipeline-processor-worker-3-thread-1] ERROR org.opensearch.dataprepper.pipeline.common.FutureHelper - FutureTask is interrupted or timed out
2023-04-03T17:19:06,287 [log-pipeline-processor-worker-3-thread-1] INFO  org.opensearch.dataprepper.pipeline.ProcessWorker - Processor shutdown phase 1 complete.
2023-04-03T17:19:07,300 [log-pipeline-processor-worker-3-thread-1] INFO  org.opensearch.dataprepper.pipeline.ProcessWorker - Beginning processor shutdown phase 2, iterating until buffers empty.
2023-04-03T17:19:07,300 [log-pipeline-sink-worker-4-thread-1] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [log-pipeline] - Shutting down process workers
Exception in thread "HTTP-Dispatcher" Exception in thread "I/O dispatcher 1" java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space
2023-04-03T17:20:00,693 [log-pipeline-sink-worker-4-thread-1] WARN  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [log-pipeline] - Workers did not terminate in time, forcing termination
2023-04-03T17:20:00,693 [metrics-pipeline-processor-worker-5-thread-1] ERROR org.opensearch.dataprepper.pipeline.common.PipelineThreadPoolExecutor - Pipeline [metrics-pipeline] process worker encountered a fatal exception, cannot proceed further
java.util.concurrent.ExecutionException: java.lang.OutOfMemoryError: Java heap space: failed reallocation of scalar replaced objects
	at java.util.concurrent.FutureTask.report(FutureTask.java:122) ~[?:?]
	at java.util.concurrent.FutureTask.get(FutureTask.java:191) ~[?:?]
	at org.opensearch.dataprepper.pipeline.common.PipelineThreadPoolExecutor.afterExecute(PipelineThreadPoolExecutor.java:70) ~[data-prepper-core-2.1.0.jar:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1137) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
	at java.lang.Thread.run(Thread.java:833) ~[?:?]
Caused by: java.lang.OutOfMemoryError: Java heap space: failed reallocation of scalar replaced objects
2023-04-03T17:20:00,693 [metrics-pipeline-processor-worker-5-thread-1] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [metrics-pipeline] - Received shutdown signal with processor shutdown timeout PT30S and sink shutdown timeout PT30S. Initiating the shutdown process
2023-04-03T17:20:00,693 [log-pipeline-processor-worker-3-thread-1] ERROR org.opensearch.dataprepper.pipeline.ProcessWorker - Encountered exception during pipeline log-pipeline processing
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@416a9cb0[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@706b5ab7[Wrapped task = org.opensearch.dataprepper.pipeline.Pipeline$$Lambda$1099/0x00000008014068d8@6dccc3d5]] rejected from org.opensearch.dataprepper.pipeline.common.PipelineThreadPoolExecutor@6f8299f9[Shutting down, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 109]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2065) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:833) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1365) ~[?:?]
	at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:134) ~[?:?]
	at org.opensearch.dataprepper.pipeline.Pipeline.lambda$publishToSinks$4(Pipeline.java:262) ~[data-prepper-core-2.1.0.jar:?]
	at org.opensearch.dataprepper.pipeline.router.DataFlowComponentRouter.route(DataFlowComponentRouter.java:45) ~[data-prepper-core-2.1.0.jar:?]
	at org.opensearch.dataprepper.pipeline.router.Router.route(Router.java:42) ~[data-prepper-core-2.1.0.jar:?]
	at org.opensearch.dataprepper.pipeline.Pipeline.publishToSinks(Pipeline.java:261) ~[data-prepper-core-2.1.0.jar:?]
	at org.opensearch.dataprepper.pipeline.ProcessWorker.postToSink(ProcessWorker.java:117) ~[data-prepper-core-2.1.0.jar:?]
	at org.opensearch.dataprepper.pipeline.ProcessWorker.doRun(ProcessWorker.java:98) ~[data-prepper-core-2.1.0.jar:?]
	at org.opensearch.dataprepper.pipeline.ProcessWorker.run(ProcessWorker.java:52) ~[data-prepper-core-2.1.0.jar:?]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
	at java.lang.Thread.run(Thread.java:833) ~[?:?]
2023-04-03T17:20:00,694 [metrics-pipeline-processor-worker-5-thread-1] INFO  org.opensearch.dataprepper.plugins.source.otelmetrics.OTelMetricsSource - Stopped otel_metrics_source.
2023-04-03T17:20:00,694 [metrics-pipeline-processor-worker-5-thread-1] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [metrics-pipeline] - Shutting down process workers
2023-04-03T17:20:00,694 [metrics-pipeline-processor-worker-5-thread-1] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [metrics-pipeline] - Encountered interruption terminating the pipeline execution, Attempting to force the termination
2023-04-03T17:20:00,694 [metrics-pipeline-processor-worker-5-thread-1] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [metrics-pipeline] - Shutting down process workers
2023-04-03T17:20:00,694 [pool-4-thread-1] ERROR org.apache.http.impl.nio.client.CloseableHttpAsyncClientBase$1 - I/O reactor terminated abnormally
org.apache.http.nio.reactor.IOReactorException: I/O dispatch worker terminated abnormally
	at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor.execute(AbstractMultiworkerIOReactor.java:359) ~[httpcore-nio-4.4.15.jar:4.4.15]
	at org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager.execute(PoolingNHttpClientConnectionManager.java:221) ~[httpasyncclient-4.1.5.jar:4.1.5]
	at org.apache.http.impl.nio.client.CloseableHttpAsyncClientBase$1.run(CloseableHttpAsyncClientBase.java:64) ~[httpasyncclient-4.1.5.jar:4.1.5]
	at java.lang.Thread.run(Thread.java:833) ~[?:?]
Caused by: java.lang.OutOfMemoryError: Java heap space

Expected behavior
PID 1 to be terminated, and then the container/pod to terminate because PID 1 (the init cmd for the container) was terminated.

I ran this after it crashed & you can see PID 1 was not terminated

data-prepper-775db9b7b6-msg2v:/usr/share/data-prepper# ps -A
PID   USER     TIME  COMMAND
    1 root      5:23 java -Ddata-prepper.dir=/usr/share/data-prepper -Dlog4j.configurationFile=/usr/share/data-prepper/config/log4j2-rolling.properties -cp /u
  482 root      0:00 sh -c clear; (bash || ash || sh)
  489 root      0:00 bash
  490 root      0:00 ps -A

Environment (please complete the following information):

• Docker
• Kubernetes
• Data prepper v2.1.0

Additional context
Note: data prepper does not crash when the resource cpu/mem limit is gone and I can view the resources used spike way over my preset limit if I run the same test.