[
https://issues.apache.org/jira/browse/BEAM-4613?focusedWorklogId=116605&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-116605
]
ASF GitHub Bot logged work on BEAM-4613:
----------------------------------------
Author: ASF GitHub Bot
Created on: 27/Jun/18 21:05
Start Date: 27/Jun/18 21:05
Worklog Time Spent: 10m
Work Description: reuvenlax closed pull request #5723: [BEAM-4613] Use
ByteBuddy to generate a coder class for a specific Schema.
URL: https://github.com/apache/beam/pull/5723
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/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoder.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoder.java
index 50a9622806e..d79fbbe8416 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoder.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoder.java
@@ -21,10 +21,10 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.BitSet;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
+import javax.annotation.Nullable;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
@@ -36,7 +36,8 @@
*/
@Experimental
public class RowCoder extends CustomCoder<Row> {
- private static final ImmutableMap<TypeName, Coder> CODER_MAP =
+ // This contains a map of primitive types to their coders.
+ static final ImmutableMap<TypeName, Coder> CODER_MAP =
ImmutableMap.<TypeName, Coder>builder()
.put(TypeName.BYTE, ByteCoder.of())
.put(TypeName.INT16, BigEndianShortCoder.of())
@@ -63,9 +64,49 @@
.put(TypeName.DATETIME, Long.BYTES)
.build();
- private static final BitSetCoder nullListCoder = BitSetCoder.of();
+ private final Schema schema;
+ private final UUID id;
+ @Nullable
+ private transient Coder<Row> delegateCoder = null;
- private Schema schema;
+ public static RowCoder of(Schema schema) {
+ return new RowCoder(schema, UUID.randomUUID());
+ }
+
+
+ private RowCoder(Schema schema, UUID id) {
+ this.schema = schema;
+ this.id = id;
+ }
+
+ // Return the generated coder class for this schema.
+ private Coder<Row> getDelegateCoder() {
+ if (delegateCoder == null) {
+ // RowCoderGenerator caches based on id, so if a new instance of this
RowCoder is
+ // deserialized, we don't need to run ByteBuddy again to construct the
class.
+ delegateCoder = RowCoderGenerator.generate(schema, id);
+ }
+ return delegateCoder;
+ }
+
+ @Override
+ public void encode(Row value, OutputStream outStream) throws IOException {
+ getDelegateCoder().encode(value, outStream);
+ }
+
+ @Override
+ public Row decode(InputStream inStream) throws IOException {
+ return getDelegateCoder().decode(inStream);
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ @Override
+ public void verifyDeterministic()
+ throws org.apache.beam.sdk.coders.Coder.NonDeterministicException {
+ }
/**
* Returns the coder used for a given primitive type.
@@ -118,78 +159,4 @@ private static long estimatedSizeBytes(FieldType
typeDescriptor, Object value) {
return ESTIMATED_FIELD_SIZES.get(typeDescriptor.getTypeName());
}
}
-
- private RowCoder(Schema schema) {
- this.schema = schema;
- }
-
- public static RowCoder of(Schema schema) {
- return new RowCoder(schema);
- }
-
- public Schema getSchema() {
- return schema;
- }
-
- Coder getCoder(FieldType fieldType) {
- if (TypeName.ARRAY.equals(fieldType.getTypeName())) {
- return ListCoder.of(getCoder(fieldType.getCollectionElementType()));
- } else if (TypeName.MAP.equals(fieldType.getTypeName())) {
- return MapCoder.of(
- coderForPrimitiveType(fieldType.getMapKeyType().getTypeName()),
- getCoder(fieldType.getMapValueType()));
- } else if (TypeName.ROW.equals((fieldType.getTypeName()))) {
- return RowCoder.of(fieldType.getRowSchema());
- } else {
- return coderForPrimitiveType(fieldType.getTypeName());
- }
- }
-
- @Override
- public void encode(Row value, OutputStream outStream) throws IOException {
- nullListCoder.encode(scanNullFields(value), outStream);
-
- for (int idx = 0; idx < value.getFieldCount(); ++idx) {
- Schema.Field field = schema.getField(idx);
- if (value.getValue(idx) == null) {
- continue;
- }
- Coder coder = getCoder(field.getType());
- coder.encode(value.getValue(idx), outStream);
- }
- }
-
- @Override
- public Row decode(InputStream inStream) throws IOException {
- BitSet nullFields = nullListCoder.decode(inStream);
- List<Object> fieldValues = new ArrayList<>(schema.getFieldCount());
- for (int idx = 0; idx < schema.getFieldCount(); ++idx) {
- if (nullFields.get(idx)) {
- fieldValues.add(null);
- } else {
- Coder coder = getCoder(schema.getField(idx).getType());
- Object value = coder.decode(inStream);
- fieldValues.add(value);
- }
- }
- return Row.withSchema(schema).addValues(fieldValues).build();
- }
-
- /**
- * Scan {@link Row} to find fields with a NULL value.
- */
- private BitSet scanNullFields(Row row) {
- BitSet nullFields = new BitSet(row.getFieldCount());
- for (int idx = 0; idx < row.getFieldCount(); ++idx) {
- if (row.getValue(idx) == null) {
- nullFields.set(idx);
- }
- }
- return nullFields;
- }
-
- @Override
- public void verifyDeterministic()
- throws org.apache.beam.sdk.coders.Coder.NonDeterministicException {
- }
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
new file mode 100644
index 00000000000..7ff3bfcb979
--- /dev/null
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
@@ -0,0 +1,370 @@
+/*
+ * 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.sdk.coders;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.reflect.InvocationTargetException;
+import java.util.BitSet;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.description.modifier.FieldManifestation;
+import net.bytebuddy.description.modifier.Ownership;
+import net.bytebuddy.description.modifier.Visibility;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.description.type.TypeDescription.ForLoadedType;
+import net.bytebuddy.dynamic.DynamicType;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.dynamic.scaffold.InstrumentedType;
+import net.bytebuddy.implementation.FixedValue;
+import net.bytebuddy.implementation.Implementation;
+import net.bytebuddy.implementation.bytecode.ByteCodeAppender;
+import net.bytebuddy.implementation.bytecode.Duplication;
+import net.bytebuddy.implementation.bytecode.StackManipulation;
+import net.bytebuddy.implementation.bytecode.StackManipulation.Compound;
+import net.bytebuddy.implementation.bytecode.TypeCreation;
+import net.bytebuddy.implementation.bytecode.collection.ArrayFactory;
+import net.bytebuddy.implementation.bytecode.member.FieldAccess;
+import net.bytebuddy.implementation.bytecode.member.MethodInvocation;
+import net.bytebuddy.implementation.bytecode.member.MethodReturn;
+import net.bytebuddy.implementation.bytecode.member.MethodVariableAccess;
+import net.bytebuddy.matcher.ElementMatchers;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.Schema.TypeName;
+import org.apache.beam.sdk.values.Row;
+
+/**
+ * A utility for automatically generating a {@link Coder} for {@link Row}
objects corresponding to a
+ * specific schema. The resulting coder is loaded into the default ClassLoader
and returned.
+ *
+ * <p>When {@link RowCoderGenerator#generate(Schema, UUID)} is called, a new
subclass of
+ * {@literal Coder<Row>} is generated for the specified schema. This class is
generated using
+ * low-level bytecode generation, and hardcodes encodings for all fields of
the Schema. Empirically,
+ * this is 30-40% faster than a coder that introspects the schema.
+ *
+ * <p>The generated class corresponds to the following Java class:
+ * <pre>{@code
+ * class SchemaRowCoder extends Coder<Row> {
+ * // Generated array containing a coder for each field in the Schema.
+ * private static final Coder[] FIELD_CODERS;
+ *
+ * // Generated method to return the schema this class corresponds to. Used
during code
+ * // generation.
+ * private static getSchema() {
+ * return schema;
+ * }
+ *
+ * {@literal @}Override public void encode(T value, OutputStream outStream) {
+ * // Delegate to a method that evaluates each coder in the static array.
+ * encodeDelegate(FIELD_CODERS, value, outStream);
+ * }
+ *
+ * {@literal @}Overide public abstract T decode(InputStream inStream) {
+ * // Delegate to a method that evaluates each coder in the static array.
+ * return decodeDelegate(FIELD_CODERS, inStream);
+ * }
+ * }
+ * }</pre>
+ */
+public abstract class RowCoderGenerator {
+ private static final ByteBuddy BYTE_BUDDY = new ByteBuddy();
+ private static final ForLoadedType CODER_TYPE = new
ForLoadedType(Coder.class);
+ private static final ForLoadedType LIST_CODER_TYPE = new
ForLoadedType(ListCoder.class);
+ private static final ForLoadedType MAP_CODER_TYPE = new
ForLoadedType(MapCoder.class);
+ private static final BitSetCoder NULL_LIST_CODER = BitSetCoder.of();
+
+ private static final String CODERS_FIELD_NAME = "FIELD_CODERS";
+
+ // A map of primitive types -> StackManipulations to create their coders.
+ private static final Map<TypeName, StackManipulation> CODER_MAP;
+
+ // Cache for Coder class that are already generated.
+ private static Map<UUID, Coder<Row>> generatedCoders = Maps.newHashMap();
+
+ static {
+ // Initialize the CODER_MAP with the StackManipulations to create the
primitive coders.
+ // Assumes that each class contains a static of() constructor method.
+ CODER_MAP = Maps.newHashMap();
+ for (Map.Entry<TypeName, Coder> entry : RowCoder.CODER_MAP.entrySet()) {
+ StackManipulation stackManipulation =
+ MethodInvocation.invoke(
+ new ForLoadedType(entry.getValue().getClass())
+ .getDeclaredMethods()
+ .filter(ElementMatchers.named(("of")))
+ .getOnly());
+ CODER_MAP.putIfAbsent(entry.getKey(), stackManipulation);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public static Coder<Row> generate(Schema schema, UUID coderId) {
+ return generatedCoders.computeIfAbsent(
+ coderId,
+ h -> {
+ TypeDescription.Generic coderType =
+ TypeDescription.Generic.Builder.parameterizedType(Coder.class,
Row.class).build();
+ DynamicType.Builder<Coder> builder =
+ (DynamicType.Builder<Coder>) BYTE_BUDDY.subclass(coderType);
+ builder = createComponentCoders(schema, builder);
+ builder = implementMethods(schema, builder);
+ try {
+ return builder
+ .make()
+ .load(Coder.class.getClassLoader(),
ClassLoadingStrategy.Default.INJECTION)
+ .getLoaded()
+ .getDeclaredConstructor()
+ .newInstance();
+ } catch (InstantiationException
+ | IllegalAccessException
+ | NoSuchMethodException
+ | InvocationTargetException e) {
+ throw new RuntimeException("Unable to generate coder for schema "
+ schema);
+ }
+ });
+ }
+
+ private static DynamicType.Builder<Coder> implementMethods(
+ Schema schema, DynamicType.Builder<Coder> builder){
+ return builder
+ .defineMethod("getSchema", Schema.class, Visibility.PRIVATE,
Ownership.STATIC)
+ .intercept(FixedValue.reference(schema))
+ .method(ElementMatchers.named("encode"))
+ .intercept(new EncodeInstruction())
+ .method(ElementMatchers.named("decode"))
+ .intercept(new DecodeInstruction());
+ }
+
+ private static class EncodeInstruction implements Implementation {
+ static final ForLoadedType LOADED_TYPE = new
ForLoadedType(EncodeInstruction.class);
+
+ @Override
+ public ByteCodeAppender appender(Target implementationTarget) {
+ return (methodVisitor, implementationContext, instrumentedMethod) -> {
+ StackManipulation manipulation =
+ new StackManipulation.Compound(
+ // Array of coders.
+ FieldAccess.forField(
+ implementationContext
+ .getInstrumentedType()
+ .getDeclaredFields()
+ .filter(ElementMatchers.named(CODERS_FIELD_NAME))
+ .getOnly())
+ .read(),
+ // Element to encode. (offset 1, as offset 0 is always "this").
+ MethodVariableAccess.REFERENCE.loadFrom(1),
+ // OutputStream.
+ MethodVariableAccess.REFERENCE.loadFrom(2),
+ // Call EncodeInstruction.encodeDelegate
+ MethodInvocation.invoke(
+ LOADED_TYPE
+ .getDeclaredMethods()
+ .filter(
+
ElementMatchers.isStatic().and(ElementMatchers.named("encodeDelegate")))
+ .getOnly()),
+ MethodReturn.VOID);
+ StackManipulation.Size size = manipulation.apply(methodVisitor,
implementationContext);
+ return new ByteCodeAppender.Size(size.getMaximalSize(),
instrumentedMethod.getStackSize());
+ };
+ }
+
+ @Override
+ public InstrumentedType prepare(InstrumentedType instrumentedType) {
+ return instrumentedType;
+ }
+
+ // The encode method of the generated Coder delegates to this method to
evaluate all of the
+ // per-field Coders.
+ @SuppressWarnings("unchecked")
+ static void encodeDelegate(Coder[] coders, Row value, OutputStream
outputStream)
+ throws IOException {
+ NULL_LIST_CODER.encode(scanNullFields(value), outputStream);
+ for (int idx = 0; idx < value.getFieldCount(); ++idx) {
+ Object fieldValue = value.getValue(idx);
+ if (value.getValue(idx) != null) {
+ coders[idx].encode(fieldValue, outputStream);
+ }
+ }
+ }
+
+ // Figure out which fields of the Row are null, and returns a BitSet. This
allows us to save
+ // on encoding each null field separately.
+ private static BitSet scanNullFields(Row row) {
+ BitSet nullFields = new BitSet(row.getFieldCount());
+ for (int idx = 0; idx < row.getFieldCount(); ++idx) {
+ if (row.getValue(idx) == null) {
+ nullFields.set(idx);
+ }
+ }
+ return nullFields;
+ }
+ }
+
+ private static class DecodeInstruction implements Implementation {
+ static final ForLoadedType LOADED_TYPE = new
ForLoadedType(DecodeInstruction.class);
+
+ @Override
+ public ByteCodeAppender appender(Target implementationTarget) {
+ return (methodVisitor, implementationContext, instrumentedMethod) -> {
+ StackManipulation manipulation =
+ new StackManipulation.Compound(
+ // Schema. Used in generation of DecodeInstruction.
+ MethodInvocation.invoke(
+ implementationContext
+ .getInstrumentedType()
+ .getDeclaredMethods()
+ .filter(ElementMatchers.named("getSchema"))
+ .getOnly()),
+ // Array of coders.
+ FieldAccess.forField(
+ implementationContext
+ .getInstrumentedType()
+ .getDeclaredFields()
+ .filter(ElementMatchers.named(CODERS_FIELD_NAME))
+ .getOnly())
+ .read(),
+ // read the InputStream. (offset 1, as offset 0 is always
"this").
+ MethodVariableAccess.REFERENCE.loadFrom(1),
+ MethodInvocation.invoke(
+ LOADED_TYPE
+ .getDeclaredMethods()
+ .filter(
+
ElementMatchers.isStatic().and(ElementMatchers.named("decodeDelegate")))
+ .getOnly()),
+ MethodReturn.REFERENCE);
+ StackManipulation.Size size = manipulation.apply(methodVisitor,
implementationContext);
+ return new ByteCodeAppender.Size(size.getMaximalSize(),
instrumentedMethod.getStackSize());
+ };
+ }
+
+ @Override
+ public InstrumentedType prepare(InstrumentedType instrumentedType) {
+ return instrumentedType;
+ }
+
+ // The decode method of the generated Coder delegates to this method to
evaluate all of the
+ // per-field Coders.
+ static Row decodeDelegate(Schema schema, Coder[] coders, InputStream
inputStream)
+ throws IOException {
+ BitSet nullFields = NULL_LIST_CODER.decode(inputStream);
+ List<Object> fieldValues = Lists.newArrayListWithCapacity(coders.length);
+ for (int i = 0; i < coders.length; ++i) {
+ if (nullFields.get(i)) {
+ fieldValues.add(null);
+ } else {
+ fieldValues.add(coders[i].decode(inputStream));
+ }
+ }
+ // We call attachValues instead of setValues. setValues validates every
element in the list
+ // is of the proper type, potentially converts to the internal type Row
stores, and copies
+ // all values. Since we assume that decode is always being called on a
previously-encoded
+ // Row, the values should already be validated and of the correct type.
So, we can save
+ // some processing by simply transferring ownership of the list to the
Row.
+ return Row.withSchema(schema).attachValues(fieldValues).build();
+ }
+ }
+
+ private static DynamicType.Builder<Coder> createComponentCoders(
+ Schema schema, DynamicType.Builder<Coder> builder) {
+ List<StackManipulation> componentCoders =
+ Lists.newArrayListWithCapacity(schema.getFieldCount());
+ for (int i = 0; i < schema.getFieldCount(); i++) {
+ componentCoders.add(getCoder(schema.getField(i).getType()));
+ }
+
+ return builder
+ // private static final Coder[] FIELD_CODERS;
+ .defineField(
+ CODERS_FIELD_NAME,
+ Coder[].class,
+ Visibility.PRIVATE,
+ Ownership.STATIC,
+ FieldManifestation.FINAL)
+ // Static initializer.
+ .initializer(
+ (methodVisitor, implementationContext, instrumentedMethod) -> {
+ StackManipulation manipulation =
+ new StackManipulation.Compound(
+ // Initialize the array of coders.
+
ArrayFactory.forType(CODER_TYPE.asGenericType()).withValues(componentCoders),
+ FieldAccess.forField(
+ implementationContext
+ .getInstrumentedType()
+ .getDeclaredFields()
+
.filter(ElementMatchers.named(CODERS_FIELD_NAME))
+ .getOnly())
+ .write());
+ StackManipulation.Size size =
+ manipulation.apply(methodVisitor, implementationContext);
+ return new ByteCodeAppender.Size(
+ size.getMaximalSize(), instrumentedMethod.getStackSize());
+ });
+ }
+
+ private static StackManipulation getCoder(Schema.FieldType fieldType) {
+ if (TypeName.ARRAY.equals(fieldType.getTypeName())) {
+ return listCoder(fieldType.getCollectionElementType());
+ } else if (TypeName.MAP.equals(fieldType.getTypeName())) {
+ return mapCoder(fieldType.getMapKeyType(), fieldType.getMapValueType());
+ } else if (TypeName.ROW.equals((fieldType.getTypeName()))) {
+ Coder<Row> nestedCoder = generate(fieldType.getRowSchema(),
UUID.randomUUID());
+ return rowCoder(nestedCoder.getClass());
+ } else {
+ return coderForPrimitiveType(fieldType.getTypeName());
+ }
+ }
+
+ private static StackManipulation listCoder(Schema.FieldType fieldType) {
+ StackManipulation componentCoder = getCoder(fieldType);
+ return new Compound(
+ componentCoder,
+ MethodInvocation.invoke(
+
LIST_CODER_TYPE.getDeclaredMethods().filter(ElementMatchers.named("of")).getOnly()));
+ }
+
+ static StackManipulation coderForPrimitiveType(Schema.TypeName typeName) {
+ return CODER_MAP.get(typeName);
+ }
+
+ static StackManipulation mapCoder(Schema.FieldType keyType, Schema.FieldType
valueType) {
+ StackManipulation keyCoder = coderForPrimitiveType(keyType.getTypeName());
+ StackManipulation valueCoder = getCoder(valueType);
+ return new Compound(
+ keyCoder,
+ valueCoder,
+ MethodInvocation.invoke(
+
MAP_CODER_TYPE.getDeclaredMethods().filter(ElementMatchers.named("of")).getOnly()));
+ }
+
+ static StackManipulation rowCoder(Class coderClass) {
+ ForLoadedType loadedType = new ForLoadedType(coderClass);
+ return new Compound(
+ TypeCreation.of(loadedType),
+ Duplication.SINGLE,
+ MethodInvocation.invoke(
+ loadedType
+ .getDeclaredMethods()
+
.filter(ElementMatchers.isConstructor().and(ElementMatchers.takesArguments(0)))
+ .getOnly()));
+ }
+}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
index 88188a36ad6..8be2ba06c37 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
@@ -356,6 +356,7 @@ public static Builder withSchema(Schema schema) {
*/
public static class Builder {
private List<Object> values = new ArrayList<>();
+ private boolean attached = false;
private Schema schema;
Builder(Schema schema) {
@@ -386,6 +387,11 @@ public Builder addArray(Object ... values) {
return this;
}
+ public Builder attachValues(List<Object> values) {
+ this.attached = true;
+ return addValues(values);
+ }
+
private List<Object> verify(Schema schema, List<Object> values) {
List<Object> verifiedValues =
Lists.newArrayListWithCapacity(values.size());
if (schema.getFieldCount() != values.size()) {
@@ -482,7 +488,7 @@ private Object verifyPrimitiveType(Object value, TypeName
type, String fieldName
}
break;
case INT16:
- if (value instanceof Short){
+ if (value instanceof Short) {
return value;
}
break;
@@ -545,7 +551,8 @@ private Instant verifyDateTime(Object value, String
fieldName) {
public Row build() {
checkNotNull(schema);
- return new AutoValue_Row(verify(schema, values), schema);
+ List<Object> values = attached ? this.values : verify(schema,
this.values);
+ return new AutoValue_Row(values, schema);
}
}
}
----------------------------------------------------------------
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: 116605)
Time Spent: 3h 10m (was: 3h)
> Improve performance of SchemaCoder
> ----------------------------------
>
> Key: BEAM-4613
> URL: https://issues.apache.org/jira/browse/BEAM-4613
> Project: Beam
> Issue Type: Sub-task
> Components: sdk-java-core
> Reporter: Reuven Lax
> Assignee: Reuven Lax
> Priority: Major
> Time Spent: 3h 10m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
