steveniemitz commented on issue #11154: [BEAM-1819] Key should be available in @OnTimer methods URL: https://github.com/apache/beam/pull/11154#issuecomment-601280654 so I think this was enough to get it working in dataflow: ``` diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java b/runners/google-cloud-dataflow-j ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java index 2c96736..96ce156 100644 --- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java +++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java @@ -373,7 +373,7 @@ public class SimpleParDoFn<InputT, OutputT> implements ParDoFn { fnRunner.onTimer( timer.getTimerId(), timer.getTimerFamilyId(), - "", + this.stepContext.stateInternals().getKey(), window, timer.getTimestamp(), timer.getOutputTimestamp(), diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java b/runners/google-cloud- dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java index 1ecdd50..04e4b15 100644 --- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java +++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java @@ -123,6 +123,7 @@ import org.apache.beam.runners.dataflow.worker.windmill.WindmillServerStub.Commi import org.apache.beam.runners.dataflow.worker.windmill.WindmillServerStub.GetWorkStream; import org.apache.beam.runners.dataflow.worker.windmill.WindmillServerStub.StreamPool; import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.extensions.gcp.util.Transport; import org.apache.beam.sdk.fn.IdGenerator; import org.apache.beam.sdk.fn.IdGenerators; @@ -1365,6 +1366,10 @@ public class StreamingDataflowWorker { // Note that TimerOrElementCoder is a backwards-compatibility class // that is really a FakeKeyedWorkItemCoder Coder<?> valueCoder = ((WindowedValueCoder<?>) readCoder).getValueCoder(); + + if (valueCoder instanceof KvCoder<?, ?>) { + return ((KvCoder<?, ?>) valueCoder).getKeyCoder(); + } if (!(valueCoder instanceof WindmillKeyedWorkItem.FakeKeyedWorkItemCoder<?, ?>)) { return null; } ``` cc @reuvenlax on the worker changes.
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
