GetData Windmill RPC calls Become Stuck

message: "Operation ongoing in step WriteParquetToGCS/ProcessEmptyWindows/CountOnWindow/Combine.perKeyWithFanout(Count)/PreCombineHot/Combine.GroupedValues for at least 17h20m00s without outputting or completing in state windmill-read at java.base@11.0.9/jdk.internal.misc.Unsafe.park(Native Method) at java.base@11.0.9/java.util.concurrent.locks.LockSupport.park(LockSupport.java:194) at java.base@11.0.9/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2081) at java.base@11.0.9/java.util.concurrent.LinkedBlockingDeque.takeFirst(LinkedBlockingDeque.java:483) at java.base@11.0.9/java.util.concurrent.LinkedBlockingDeque.take(LinkedBlockingDeque.java:671) at app//org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$AppendableInputStream$1.hasMoreElements(GrpcWindmillServer.java:1527) at java.base@11.0.9/java.io.SequenceInputStream.peekNextStream(SequenceInputStream.java:101) at java.base@11.0.9/java.io.SequenceInputStream.nextStream(SequenceInputStream.java:97) at java.base@11.0.9/java.io.SequenceInputStream.read(SequenceInputStream.java:203) at app//org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$AppendableInputStream.read(GrpcWindmillServer.java:1603) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.CodedInputStream$StreamDecoder.tryRefillBuffer(CodedInputStream.java:2786) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.CodedInputStream$StreamDecoder.isAtEnd(CodedInputStream.java:2709) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.CodedInputStream$StreamDecoder.readTag(CodedInputStream.java:2060) at app//org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse.<init>(Windmill.java:37549) at app//org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse.<init>(Windmill.java:37505) at app//org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse$1.parsePartialFrom(Windmill.java:40068) at app//org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse$1.parsePartialFrom(Windmill.java:40062) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:215) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:232) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:237) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:48) at app//org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.GeneratedMessageV3.parseWithIOException(GeneratedMessageV3.java:339) at app//org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse.parseFrom(Windmill.java:38173) at app//org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$GrpcGetDataStream$$Lambda$342/0x0000000800746040.parse(Unknown Source) at app//org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$GrpcGetDataStream.issueRequest(GrpcWindmillServer.java:1163) at app//org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$GrpcGetDataStream.requestKeyedData(GrpcWindmillServer.java:1116) at app//org.apache.beam.runners.dataflow.worker.MetricTrackingWindmillServerStub.getStateData(MetricTrackingWindmillServerStub.java:220) at app//org.apache.beam.runners.dataflow.worker.WindmillStateReader.startBatchAndBlock(WindmillStateReader.java:474) at app//org.apache.beam.runners.dataflow.worker.WindmillStateReader$WrappedFuture.get(WindmillStateReader.java:365) at app//org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillWatermarkHold.read(WindmillStateInternals.java:1847) at app//org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillWatermarkHold.read(WindmillStateInternals.java:1794) at app//org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.WatermarkHold$1.read(WatermarkHold.java:414) at app//org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.WatermarkHold$1.read(WatermarkHold.java:403) at app//org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.onTrigger(ReduceFnRunner.java:994) at app//org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.onTimers(ReduceFnRunner.java:771) at app//org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:97) at app//org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:43) at app//org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.invokeProcessElement(GroupAlsoByWindowFnRunner.java:121) at app//org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.processElement(GroupAlsoByWindowFnRunner.java:73) at app//org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.LateDataDroppingDoFnRunner.processElement(LateDataDroppingDoFnRunner.java:80) at app//org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn.processElement(GroupAlsoByWindowsParDoFn.java:137) at app//org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44) at app//org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49) at app//org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:212) at app//org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:163) at app//org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:92) at app//org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1432) at app//org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$1100(StreamingDataflowWorker.java:165) at app//org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$7.run(StreamingDataflowWorker.java:1111) at java.base@11.0.9/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base@11.0.9/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base@11.0.9/java.lang.Thread.run(Thread.java:834)