Remove cancelled tasks from ReadOperation queue when shutting down (#34335)

* Remove cancelled tasks from ReadOperation queue when shutting down

* Try shutdowNow

* spotless

Author: damccorm

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

@@ -225,8 +225,16 @@ public void run() {
         if (!scheduler.isTerminated()) {
           LOG.error(
               "Failed to terminate periodic progress reporting in 1 minute. "
-                  + "Waiting for it to terminate indefinitely...");
-          scheduler.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
+                  + "Waiting for it to terminate 10 minutes before forcing");
+          scheduler.awaitTermination(10, TimeUnit.MINUTES);
+          if (!scheduler.isTerminated()) {
+            LOG.error(
+                "Failed to terminate periodic progress reporting in 10 "
+                    + "minutes. Trying to force termination then waiting "
+                    + "indefinitely...");
+            scheduler.shutdownNow();
+            scheduler.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
+          }
           LOG.info("Periodic progress reporting terminated.");
         }
       }