[
https://issues.apache.org/jira/browse/BEAM-4659?focusedWorklogId=116985&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-116985
]
ASF GitHub Bot logged work on BEAM-4659:
----------------------------------------
Author: ASF GitHub Bot
Created on: 28/Jun/18 19:56
Start Date: 28/Jun/18 19:56
Worklog Time Spent: 10m
Work Description: lukecwik closed pull request #5794: [BEAM-4659] Add
well known timer coder to Java SDK.
URL: https://github.com/apache/beam/pull/5794
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml
b/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml
index 494e749a1df..f7ede2b0630 100644
---
a/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml
+++
b/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml
@@ -152,6 +152,20 @@ examples:
---
+coder:
+ urn: "beam:coder:timer:v1"
+ components: [{urn: "beam:coder:bytes:v1"}]
+examples:
+ "\0\0\0\0\0\0\0\0\u0003abc": {timestamp: -9223372036854775808, payload: abc}
+ "\x7fÿÿÿÿÿÿ\x01\u0003abc": {timestamp: -255, payload: abc}
+ "\x7fÿÿÿÿÿÿÿ\u0003abc": {timestamp: -1, payload: abc}
+ "\x80\0\0\0\0\0\0\0\u0003abc": {timestamp: 0, payload: abc}
+ "\x80\0\0\0\0\0\0\x01\u0003abc": {timestamp: 1, payload: abc}
+ "\x80\0\0\0\0\0\x01\0\u0003abc": {timestamp: 256, payload: abc}
+ "ÿÿÿÿÿÿÿÿ\u0003abc": {timestamp: 9223372036854775807, payload: abc}
+
+---
+
coder:
urn: "beam:coder:global_window:v1"
examples:
diff --git
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/CoderTranslators.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/CoderTranslators.java
index ce3d11c5355..7481c2da57a 100644
---
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/CoderTranslators.java
+++
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/CoderTranslators.java
@@ -76,6 +76,20 @@ public T fromComponents(List<Coder<?>> components) {
};
}
+ static CoderTranslator<Timer.Coder<?>> timer() {
+ return new SimpleStructuredCoderTranslator<Timer.Coder<?>>() {
+ @Override
+ public List<? extends Coder<?>> getComponents(Timer.Coder<?> from) {
+ return from.getCoderArguments();
+ }
+
+ @Override
+ public Timer.Coder<?> fromComponents(List<Coder<?>> components) {
+ return Timer.Coder.of(components.get(0));
+ }
+ };
+ }
+
static CoderTranslator<LengthPrefixCoder<?>> lengthPrefix() {
return new SimpleStructuredCoderTranslator<LengthPrefixCoder<?>>() {
@Override
diff --git
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoderRegistrar.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoderRegistrar.java
index 2ddd053a9e0..424889807e5 100644
---
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoderRegistrar.java
+++
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoderRegistrar.java
@@ -51,6 +51,7 @@
.put(VarLongCoder.class, ModelCoders.INT64_CODER_URN)
.put(IntervalWindowCoder.class,
ModelCoders.INTERVAL_WINDOW_CODER_URN)
.put(IterableCoder.class, ModelCoders.ITERABLE_CODER_URN)
+ .put(Timer.Coder.class, ModelCoders.TIMER_CODER_URN)
.put(LengthPrefixCoder.class, ModelCoders.LENGTH_PREFIX_CODER_URN)
.put(GlobalWindow.Coder.class, ModelCoders.GLOBAL_WINDOW_CODER_URN)
.put(FullWindowedValueCoder.class,
ModelCoders.WINDOWED_VALUE_CODER_URN)
@@ -67,6 +68,7 @@
.put(GlobalWindow.Coder.class,
CoderTranslators.atomic(GlobalWindow.Coder.class))
.put(KvCoder.class, CoderTranslators.kv())
.put(IterableCoder.class, CoderTranslators.iterable())
+ .put(Timer.Coder.class, CoderTranslators.timer())
.put(LengthPrefixCoder.class, CoderTranslators.lengthPrefix())
.put(FullWindowedValueCoder.class,
CoderTranslators.fullWindowedValue())
.build();
diff --git
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoders.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoders.java
index 3544d66d850..fd19a4d3cb9 100644
---
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoders.java
+++
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ModelCoders.java
@@ -39,6 +39,7 @@ private ModelCoders() {}
public static final String INT64_CODER_URN =
getUrn(StandardCoders.Enum.VARINT);
public static final String ITERABLE_CODER_URN =
getUrn(StandardCoders.Enum.ITERABLE);
+ public static final String TIMER_CODER_URN =
getUrn(StandardCoders.Enum.TIMER);
public static final String KV_CODER_URN = getUrn(StandardCoders.Enum.KV);
public static final String LENGTH_PREFIX_CODER_URN =
getUrn(StandardCoders.Enum.LENGTH_PREFIX);
@@ -56,6 +57,7 @@ private ModelCoders() {}
BYTES_CODER_URN,
INT64_CODER_URN,
ITERABLE_CODER_URN,
+ TIMER_CODER_URN,
KV_CODER_URN,
LENGTH_PREFIX_CODER_URN,
GLOBAL_WINDOW_CODER_URN,
diff --git
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/Timer.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/Timer.java
new file mode 100644
index 00000000000..77962815b04
--- /dev/null
+++
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/Timer.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core.construction;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.InstantCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+import org.joda.time.Instant;
+
+/**
+ * A timer consists of a timestamp and a corresponding user supplied payload.
+ *
+ * <p>Note that this implementation is specifically used during execution
within runners and inside
+ * the SDK harness. The {@link org.apache.beam.sdk.state.Timer} represents the
current user facing
+ * API. Consider consolidating the two once {@link
org.apache.beam.runners.core.TimerInternals} is
+ * no longer the way in which users construct/interact with timers.
+ */
+@AutoValue
+public abstract class Timer<T> {
+
+ /** Returns a timer for the given timestamp with a {@code null} payload. */
+ public static Timer<Void> of(Instant time) {
+ return of(time, (Void) null);
+ }
+
+ /** Returns a timer for the given timestamp with a user specified payload. */
+ public static <T> Timer<T> of(Instant timestamp, @Nullable T payload) {
+ return new AutoValue_Timer(timestamp, payload);
+ }
+
+ /**
+ * Returns the timestamp of when the timer is scheduled to fire.
+ *
+ * <p>The time is relative to the time domain defined in the {@link
+ * org.apache.beam.model.pipeline.v1.RunnerApi.TimerSpec} that is associated
with this timer.
+ */
+ public abstract Instant getTimestamp();
+
+ /** A user supplied payload. */
+ @Nullable
+ public abstract T getPayload();
+
+ /**
+ * A {@link org.apache.beam.sdk.coders.Coder} for timers.
+ *
+ * <p>This coder is deterministic if the payload coder is deterministic.
+ *
+ * <p>This coder is inexpensive for size estimation of elements if the
payload coder is
+ * inexpensive for size estimation.
+ */
+ public static class Coder<T> extends StructuredCoder<Timer<T>> {
+
+ public static <T> Coder of(org.apache.beam.sdk.coders.Coder<T>
payloadCoder) {
+ return new Coder(payloadCoder);
+ }
+
+ private final org.apache.beam.sdk.coders.Coder<T> payloadCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<T> payloadCoder) {
+ this.payloadCoder = payloadCoder;
+ }
+
+ @Override
+ public void encode(Timer<T> timer, OutputStream outStream) throws
CoderException, IOException {
+ InstantCoder.of().encode(timer.getTimestamp(), outStream);
+ payloadCoder.encode(timer.getPayload(), outStream);
+ }
+
+ @Override
+ public Timer<T> decode(InputStream inStream) throws CoderException,
IOException {
+ Instant instant = InstantCoder.of().decode(inStream);
+ T value = payloadCoder.decode(inStream);
+ return Timer.of(instant, value);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>>
getCoderArguments() {
+ return Collections.singletonList(payloadCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ verifyDeterministic(this, "Payload coder must be deterministic",
payloadCoder);
+ }
+
+ @Override
+ public boolean consistentWithEquals() {
+ return payloadCoder.consistentWithEquals();
+ }
+
+ @Override
+ public Object structuralValue(Timer<T> value) {
+ return Timer.of(value.getTimestamp(),
payloadCoder.structuralValue(value.getPayload()));
+ }
+
+ @Override
+ public boolean isRegisterByteSizeObserverCheap(Timer<T> value) {
+ return payloadCoder.isRegisterByteSizeObserverCheap(value.getPayload());
+ }
+
+ @Override
+ public void registerByteSizeObserver(Timer<T> value,
ElementByteSizeObserver observer)
+ throws Exception {
+ InstantCoder.of().registerByteSizeObserver(value.getTimestamp(),
observer);
+ payloadCoder.registerByteSizeObserver(value.getPayload(), observer);
+ }
+ }
+}
diff --git
a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CoderTranslationTest.java
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CoderTranslationTest.java
index 8565fc59380..514e4a22e15 100644
---
a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CoderTranslationTest.java
+++
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CoderTranslationTest.java
@@ -65,6 +65,7 @@
.add(VarLongCoder.of())
.add(IntervalWindowCoder.of())
.add(IterableCoder.of(ByteArrayCoder.of()))
+ .add(Timer.Coder.of(ByteArrayCoder.of()))
.add(LengthPrefixCoder.of(IterableCoder.of(VarLongCoder.of())))
.add(GlobalWindow.Coder.INSTANCE)
.add(
diff --git
a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java
index 8688a726c74..7d097bc820b 100644
---
a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java
+++
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java
@@ -88,6 +88,7 @@
.put(getUrn(StandardCoders.Enum.VARINT), VarLongCoder.class)
.put(getUrn(StandardCoders.Enum.INTERVAL_WINDOW),
IntervalWindowCoder.class)
.put(getUrn(StandardCoders.Enum.ITERABLE), IterableCoder.class)
+ .put(getUrn(StandardCoders.Enum.TIMER), Timer.Coder.class)
.put(getUrn(StandardCoders.Enum.GLOBAL_WINDOW),
GlobalWindow.Coder.class)
.put(
getUrn(StandardCoders.Enum.WINDOWED_VALUE),
@@ -226,6 +227,12 @@ private static Object convertValue(Object value,
CommonCoder coderSpec, Coder co
return KV.of(k, v);
} else if (s.equals(getUrn(StandardCoders.Enum.VARINT))) {
return ((Number) value).longValue();
+ } else if (s.equals(getUrn(StandardCoders.Enum.TIMER))) {
+ Map<String, Object> kvMap = (Map<String, Object>) value;
+ Coder<?> payloadCoder = (Coder) coder.getCoderArguments().get(0);
+ return Timer.of(
+ new Instant(((Number) kvMap.get("timestamp")).longValue()),
+ convertValue(kvMap.get("payload"), coderSpec.getComponents().get(0),
payloadCoder));
} else if (s.equals(getUrn(StandardCoders.Enum.INTERVAL_WINDOW))) {
Map<String, Object> kvMap = (Map<String, Object>) value;
Instant end = new Instant(((Number) kvMap.get("end")).longValue());
@@ -284,6 +291,8 @@ private static Object convertValue(Object value,
CommonCoder coderSpec, Coder co
return IntervalWindowCoder.of();
} else if (s.equals(getUrn(StandardCoders.Enum.ITERABLE))) {
return IterableCoder.of(components.get(0));
+ } else if (s.equals(getUrn(StandardCoders.Enum.TIMER))) {
+ return Timer.Coder.of(components.get(0));
} else if (s.equals(getUrn(StandardCoders.Enum.GLOBAL_WINDOW))) {
return GlobalWindow.Coder.INSTANCE;
} else if (s.equals(getUrn(StandardCoders.Enum.WINDOWED_VALUE))) {
@@ -338,6 +347,10 @@ private void verifyDecodedValue(CommonCoder coder, Object
expectedValue, Object
}
assertFalse(expectedValueIterator.hasNext());
+ } else if (s.equals(getUrn(StandardCoders.Enum.TIMER))) {
+ assertEquals(((Timer) expectedValue).getTimestamp(), ((Timer)
actualValue).getTimestamp());
+ assertThat(((Timer) expectedValue).getPayload(), equalTo(((Timer)
actualValue).getPayload()));
+
} else if (s.equals(getUrn(StandardCoders.Enum.GLOBAL_WINDOW))) {
assertEquals(expectedValue, actualValue);
diff --git
a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/TimerTest.java
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/TimerTest.java
new file mode 100644
index 00000000000..ac3bd70c981
--- /dev/null
+++
b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/TimerTest.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core.construction;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.testing.CoderProperties;
+import org.joda.time.Instant;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link Timer}. */
+@RunWith(JUnit4.class)
+public class TimerTest {
+ private static final Instant INSTANT = Instant.now();
+
+ @Test
+ public void testTimer() {
+ Timer<Void> timerA = Timer.of(INSTANT);
+ assertEquals(INSTANT, timerA.getTimestamp());
+ assertNull(timerA.getPayload());
+
+ Timer<String> timerB = Timer.of(INSTANT, "ABC");
+ assertEquals(INSTANT, timerB.getTimestamp());
+ assertEquals("ABC", timerB.getPayload());
+ }
+
+ @Test
+ public void testTimerCoderWithInconsistentWithEqualsPayloadCoder() throws
Exception {
+ Coder<Timer<byte[]>> coder = Timer.Coder.of(ByteArrayCoder.of());
+ CoderProperties.coderSerializable(coder);
+ CoderProperties.structuralValueDecodeEncodeEqual(
+ coder, Timer.of(INSTANT, "ABC".getBytes(UTF_8)));
+ CoderProperties.structuralValueConsistentWithEquals(
+ coder, Timer.of(INSTANT, "ABC".getBytes(UTF_8)), Timer.of(INSTANT,
"ABC".getBytes(UTF_8)));
+ }
+
+ @Test
+ public void testTimerCoderWithConsistentWithEqualsPayloadCoder() throws
Exception {
+ Coder<Timer<String>> coder = Timer.Coder.of(StringUtf8Coder.of());
+ CoderProperties.coderDecodeEncodeEqual(coder, Timer.of(INSTANT, "ABC"));
+ CoderProperties.coderConsistentWithEquals(
+ coder, Timer.of(INSTANT, "ABC"), Timer.of(INSTANT, "ABC"));
+ CoderProperties.coderDeterministic(coder, Timer.of(INSTANT, "ABC"),
Timer.of(INSTANT, "ABC"));
+ }
+
+ @Test
+ public void testTimerCoderWireFormat() throws Exception {
+ Coder<Timer<String>> coder = Timer.Coder.of(StringUtf8Coder.of());
+ CoderProperties.coderEncodesBase64(
+ coder, Timer.of(new Instant(255L), "ABC"), "gAAAAAAAAP8DQUJD");
+ }
+}
diff --git a/sdks/python/apache_beam/testing/data/standard_coders.yaml
b/sdks/python/apache_beam/testing/data/standard_coders.yaml
index ae034e521fe..494e749a1df 100644
--- a/sdks/python/apache_beam/testing/data/standard_coders.yaml
+++ b/sdks/python/apache_beam/testing/data/standard_coders.yaml
@@ -37,7 +37,6 @@
# If a coder is marked non-deterministic in the coder spec, then only the
decoding should be validated.
-
coder:
urn: "beam:coder:bytes:v1"
nested: false
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 116985)
Time Spent: 1h (was: 50m)
> Add well known timer coder for Java SDK
> ---------------------------------------
>
> Key: BEAM-4659
> URL: https://issues.apache.org/jira/browse/BEAM-4659
> Project: Beam
> Issue Type: Sub-task
> Components: sdk-java-core
> Reporter: Luke Cwik
> Assignee: Luke Cwik
> Priority: Major
> Labels: portability
> Time Spent: 1h
> Remaining Estimate: 0h
>
> Encoding is: [Instant, Payload]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
