Fix Dataflow legacy worker abort loop thread death issue

Previously, when the service asked the worker to abort, it threw
ReadLoopAbortedException, which extended InterruptedException.
MapTaskExecutor caught this and, in an attempt to preserve the
interrupted status, set the interrupted bit on the thread.
However, since this was a logical abort and not a real thread interrupt,
setting the interrupted bit caused subsequent operations on the thread
(like the backoff sleep in DataflowBatchWorkerHarness) to immediately
fail with InterruptedException, leading to all worker threads dying
and the harness hanging.

This fix:
1. Changes ReadLoopAbortedException to extend InterruptedIOException
   instead of InterruptedException, as it is a logical I/O abort and
   should not trigger thread interrupt handling.
2. Hardens DataflowBatchWorkerHarness to throw a RuntimeException
   if all worker threads die, ensuring the JVM exits with a non-zero
   code so the runner can restart it.

Ref: b/512366613

Author: damccorm

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowBatchWorkerHarness.java

@@ -92,6 +92,7 @@ public void run() throws InterruptedException {
 
     worker.startStatusServer();
     processWork(pipelineOptions, worker, Sleeper.DEFAULT);
+    throw new RuntimeException("All worker threads have died. Exiting.");
   }
 
   // ExponentialBackOff parameters for the task retry strategy.

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/ReadOperation.java

@@ -19,6 +19,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.util.Observable;
 import java.util.Observer;
 import java.util.concurrent.CancellationException;
@@ -47,7 +48,7 @@
   "nullness" // TODO(https://github.com/apache/beam/issues/20497)
 })
 public class ReadOperation extends Operation {
-  public static class ReadLoopAbortedException extends InterruptedException {
+  public static class ReadLoopAbortedException extends InterruptedIOException {
     public ReadLoopAbortedException() {
       super("Read loop was aborted.");
     }