Unable to read value from state/Unable to fetch data due to token mismatch for key

[Mohil Khare]

Hello,

I am using beam java sdk 2.19.0 (with enableStreamingEngine set as true)
and very heavily use stateful beam processing model.
However, sometimes I am seeing the following exception while reading value
from state for a key (Please note : here my key is a POJO where fields
create a kind of composite key. Also I am using AvroCoder for this key):

Caused by: java.util.concurrent.ExecutionException:
org.apache.beam.runners.dataflow.worker.KeyTokenInvalidException: Unable to
fetch data due to token mismatch for key
0ggadot_static@prosimo.ioHaa552bec-25f2-11ea-8705-267acc424a25H9219bdd5-335f-11ea-bd4f-de07a30b09ca
@ OC-AU sydney

   1.
      1. at
      org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.
      AbstractFuture.getDoneValue (AbstractFuture.java:531
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent%2FAbstractFuture.java&line=531&project=prosimo>
      )
      2. at
      org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.
      AbstractFuture.get (AbstractFuture.java:492
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent%2FAbstractFuture.java&line=492&project=prosimo>
      )
      3. at
      org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.
      AbstractFuture$TrustedFuture.get (AbstractFuture.java:83
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent%2FAbstractFuture.java&line=83&project=prosimo>
      )
      4. at
      org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.
      ForwardingFuture.get (ForwardingFuture.java:62
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent%2FForwardingFuture.java&line=62&project=prosimo>
      )
      5. atorg.apache.beam.runners.dataflow.worker.
      WindmillStateReader$WrappedFuture.get (WindmillStateReader.java:316
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.runners.dataflow.worker%2FWindmillStateReader.java&line=316&project=prosimo>
      )
      6. atorg.apache.beam.runners.dataflow.worker.
      WindmillStateInternals$WindmillValue.read (
      WindmillStateInternals.java:385
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.runners.dataflow.worker%2FWindmillStateInternals.java&line=385&project=prosimo>
      )

Caused by: org.apache.beam.runners.dataflow.worker.KeyTokenInvalidException:
Unable to fetch data due to token mismatch for key <key>

   1.
      1. atorg.apache.beam.runners.dataflow.worker.
      WindmillStateReader.consumeResponse (WindmillStateReader.java:482
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.runners.dataflow.worker%2FWindmillStateReader.java&line=482&project=prosimo>
      )
      2. atorg.apache.beam.runners.dataflow.worker.
      WindmillStateReader.startBatchAndBlock (WindmillStateReader.java:420
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.runners.dataflow.worker%2FWindmillStateReader.java&line=420&project=prosimo>
      )
      3. atorg.apache.beam.runners.dataflow.worker.
      WindmillStateReader$WrappedFuture.get (WindmillStateReader.java:313
      
<https://console.cloud.google.com/debug/fromlog?appModule=Dataflow%20Jobs&appVersion=2020-07-02_10_32_35-16490648164981155775&file=org%2Fapache.beam.runners.dataflow.worker%2FWindmillStateReader.java&line=313&project=prosimo>
      )


Any help to fix this issue would be greatly appreciated.

Thanks and Regards
Mohil