[Bug]: SpannerIO.readChangeStream() stops forwarding change records and starts continuously throwing (large number) of Operation ongoing errors

### What happened?

Using a new [Dataflow template](https://github.com/GoogleCloudPlatform/DataflowTemplates/pull/429) that reads Spanner change streams via the SpannerIO.readChangeStream(). The streaming template was working correctly for 2 days but then it stops forwarding change records and starts continuously throwing errors like below

```
Operation ongoing in step SpannerIO-ReadChangeStream-Read-change-stream-partition-ParMultiDo-ReadChangeStreamPartition-/ProcessElementAndRestrictionWithSizing-ptransform-60 for at least 332h25m00s without outputting or completing in state process
at sun.misc.Unsafe.park(Native Method)
at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:338)
at com.lmax.disruptor.MultiProducerSequencer.next(MultiProducerSequencer.java:136)
at com.lmax.disruptor.MultiProducerSequencer.next(MultiProducerSequencer.java:105)
at com.lmax.disruptor.RingBuffer.next(RingBuffer.java:263)
at io.opencensus.impl.internal.DisruptorEventQueue$1.enqueue(DisruptorEventQueue.java:134)
at io.opencensus.impl.internal.DisruptorEventQueue.enqueue(DisruptorEventQueue.java:162)
at io.opencensus.implcore.stats.StatsManager.record(StatsManager.java:70)
at io.opencensus.implcore.stats.MeasureMapImpl.record(MeasureMapImpl.java:82)
at io.grpc.census.CensusStatsModule$CallAttemptsTracerFactory.(CensusStatsModule.java:447)
at io.grpc.census.CensusStatsModule$StatsClientInterceptor.interceptCall(CensusStatsModule.java:791)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.api.gax.grpc.GrpcChannelUUIDInterceptor.interceptCall(GrpcChannelUUIDInterceptor.java:52)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.api.gax.grpc.GrpcHeaderInterceptor.interceptCall(GrpcHeaderInterceptor.java:80)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.api.gax.grpc.GrpcMetadataHandlerInterceptor.interceptCall(GrpcMetadataHandlerInterceptor.java:54)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.cloud.spanner.spi.v1.SpannerErrorInterceptor.interceptCall(SpannerErrorInterceptor.java:64)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.cloud.spanner.spi.v1.LoggingInterceptor.interceptCall(LoggingInterceptor.java:68)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.cloud.spanner.spi.v1.HeaderInterceptor.interceptCall(HeaderInterceptor.java:72)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at io.grpc.internal.ManagedChannelImpl.newCall(ManagedChannelImpl.java:923)
at io.grpc.internal.ForwardingManagedChannel.newCall(ForwardingManagedChannel.java:63)
at io.grpc.stub.MetadataUtils$HeaderAttachingClientInterceptor.interceptCall(MetadataUtils.java:81)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.api.gax.grpc.GrpcClientCalls.newCall(GrpcClientCalls.java:99)
at com.google.api.gax.grpc.GrpcDirectServerStreamingCallable.call(GrpcDirectServerStreamingCallable.java:65)
at com.google.api.gax.grpc.GrpcServerStreamingRequestParamCallable.call(GrpcServerStreamingRequestParamCallable.java:61)
at com.google.api.gax.grpc.GrpcExceptionServerStreamingCallable.call(GrpcExceptionServerStreamingCallable.java:59)
at com.google.api.gax.rpc.WatchdogServerStreamingCallable.call(WatchdogServerStreamingCallable.java:69)
at com.google.api.gax.rpc.ServerStreamingCallable$1.call(ServerStreamingCallable.java:237)
at com.google.api.gax.rpc.ServerStreamingAttemptCallable.call(ServerStreamingAttemptCallable.java:234)
at com.google.api.gax.rpc.ServerStreamingAttemptCallable.start(ServerStreamingAttemptCallable.java:194)
at com.google.api.gax.rpc.RetryingServerStreamingCallable.call(RetryingServerStreamingCallable.java:87)
at com.google.api.gax.tracing.TracedServerStreamingCallable.call(TracedServerStreamingCallable.java:76)
at com.google.api.gax.rpc.ServerStreamingCallable$1.call(ServerStreamingCallable.java:237)
at com.google.cloud.spanner.spi.v1.GapicSpannerRpc.executeQuery(GapicSpannerRpc.java:1506)
at com.google.cloud.spanner.AbstractReadContext$1.startStream(AbstractReadContext.java:667)
at com.google.cloud.spanner.AbstractResultSet$ResumableStreamIterator.computeNext(AbstractResultSet.java:1105)
at com.google.cloud.spanner.AbstractResultSet$ResumableStreamIterator.computeNext(AbstractResultSet.java:986)
at com.google.common.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:146)
at com.google.common.collect.AbstractIterator.hasNext(AbstractIterator.java:141)
at com.google.cloud.spanner.AbstractResultSet$GrpcValueIterator.ensureReady(AbstractResultSet.java:269)
at com.google.cloud.spanner.AbstractResultSet$GrpcValueIterator.getMetadata(AbstractResultSet.java:245)
at com.google.cloud.spanner.AbstractResultSet$GrpcResultSet.next(AbstractResultSet.java:119)
at com.google.cloud.spanner.ForwardingResultSet.next(ForwardingResultSet.java:54)
at com.google.cloud.spanner.SessionPool$AutoClosingReadContext$1.internalNext(SessionPool.java:272)
at com.google.cloud.spanner.SessionPool$AutoClosingReadContext$1.next(SessionPool.java:252)
at org.apache.beam.sdk.io.gcp.spanner.changestreams.dao.PartitionMetadataDao.getPartition(PartitionMetadataDao.java:110)
at org.apache.beam.sdk.io.gcp.spanner.changestreams.action.QueryChangeStreamAction.run(QueryChangeStreamAction.java:166)
at org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFn.processElement(ReadChangeStreamPartitionDoFn.java:234)
at org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFn$DoFnInvoker.invokeProcessElement(Unknown Source)
at org.apache.beam.fn.harness.FnApiDoFnRunner.processElementForWindowObservingSizedElementAndRestriction(FnApiDoFnRunner.java:1063)
at org.apache.beam.fn.harness.FnApiDoFnRunner.access$1500(FnApiDoFnRunner.java:142)
at org.apache.beam.fn.harness.FnApiDoFnRunner$4.accept(FnApiDoFnRunner.java:661)
at org.apache.beam.fn.harness.FnApiDoFnRunner$4.accept(FnApiDoFnRunner.java:656)
at org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:255)
at org.apache.beam.fn.harness.data.PCollectionConsumerRegistry$MetricTrackingFnDataReceiver.accept(PCollectionConsumerRegistry.java:209)
at org.apache.beam.fn.harness.BeamFnDataReadRunner.forwardElementToConsumer(BeamFnDataReadRunner.java:179)
at org.apache.beam.fn.harness.BeamFnDataReadRunner$Factory$$Lambda$242/882597230.accept(Unknown Source)
at org.apache.beam.sdk.fn.data.BeamFnDataInboundObserver2.multiplexElements(BeamFnDataInboundObserver2.java:158)
at org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:515)
at org.apache.beam.fn.harness.FnHarness$$Lambda$108/1459016715.apply(Unknown Source)
at org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:151)
at org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver.lambda$onNext$0(BeamFnControlClient.java:116)
at org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver$$Lambda$117/539047905.run(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
```

### Issue Priority

Priority: 1

### Issue Component

Component: io-java-gcp

Start with ReadChangeStreamPartitionDoFn.java and QueryChangeStreamAction.java, following the stack trace into PartitionMetadataDao.java at getPartition. Reproduce the long-running Dataflow change-stream failure if possible and determine why records stop forwarding while Operation ongoing errors repeat; done means the stream continues forwarding records without the repeated errors.

Written by the indexing model from the issue text.