You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/11/18 01:23:48 UTC
[1/4] beam git commit: This closes #4133
Repository: beam
Updated Branches:
refs/heads/master 50e58f0bb -> e3c1fe4b1
This closes #4133
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/e3c1fe4b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/e3c1fe4b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/e3c1fe4b
Branch: refs/heads/master
Commit: e3c1fe4b157375f12b2d774af2928eedcd7dfba4
Parents: 50e58f0 8101103
Author: Thomas Groh <tg...@google.com>
Authored: Fri Nov 17 17:23:41 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Nov 17 17:23:41 2017 -0800
----------------------------------------------------------------------
pom.xml | 6 +
runners/direct-java/pom.xml | 6 +
.../beam/runners/direct/BundleFactory.java | 1 +
.../runners/direct/CloningBundleFactory.java | 1 +
.../beam/runners/direct/CommittedBundle.java | 10 +-
.../runners/direct/DirectExecutionContext.java | 1 +
.../runners/direct/DirectTransformExecutor.java | 196 +++++++
.../beam/runners/direct/EvaluationContext.java | 1 +
.../direct/ExecutorServiceParallelExecutor.java | 56 +-
.../GroupAlsoByWindowEvaluatorFactory.java | 1 +
.../direct/GroupByKeyOnlyEvaluatorFactory.java | 1 +
.../ImmutabilityCheckingBundleFactory.java | 1 +
.../direct/ImmutableListBundleFactory.java | 9 +
.../beam/runners/direct/MultiStepCombine.java | 1 +
.../beam/runners/direct/ParDoEvaluator.java | 1 +
.../runners/direct/ParDoEvaluatorFactory.java | 1 +
.../direct/StatefulParDoEvaluatorFactory.java | 1 +
.../apache/beam/runners/direct/StepAndKey.java | 1 +
.../beam/runners/direct/StructuralKey.java | 103 ----
.../beam/runners/direct/TransformExecutor.java | 165 +-----
.../direct/TransformExecutorFactory.java | 32 ++
.../direct/TransformExecutorService.java | 6 +-
.../direct/TransformExecutorServices.java | 14 +-
.../direct/UnboundedReadDeduplicator.java | 1 +
.../beam/runners/direct/WatermarkManager.java | 5 +-
.../direct/CloningBundleFactoryTest.java | 1 +
.../direct/DirectRunnerApiSurfaceTest.java | 5 +-
.../direct/DirectTimerInternalsTest.java | 1 +
.../direct/DirectTransformExecutorTest.java | 537 +++++++++++++++++++
.../runners/direct/EvaluationContextTest.java | 1 +
.../direct/GroupByKeyEvaluatorFactoryTest.java | 1 +
.../GroupByKeyOnlyEvaluatorFactoryTest.java | 1 +
.../ImmutabilityCheckingBundleFactoryTest.java | 1 +
.../direct/ImmutableListBundleFactoryTest.java | 5 +-
.../beam/runners/direct/ParDoEvaluatorTest.java | 1 +
.../StatefulParDoEvaluatorFactoryTest.java | 1 +
.../beam/runners/direct/StructuralKeyTest.java | 1 +
.../direct/TransformExecutorServicesTest.java | 40 +-
.../runners/direct/TransformExecutorTest.java | 537 -------------------
.../runners/direct/WatermarkManagerTest.java | 1 +
runners/local-java/pom.xml | 75 +++
.../org/apache/beam/runners/local/Bundle.java | 33 ++
.../beam/runners/local/StructuralKey.java | 103 ++++
.../apache/beam/runners/local/package-info.java | 22 +
.../beam/runners/local/StructuralKeyTest.java | 88 +++
runners/pom.xml | 1 +
46 files changed, 1200 insertions(+), 877 deletions(-)
----------------------------------------------------------------------
[4/4] beam git commit: Add a Local Java Core Module
Posted by tg...@apache.org.
Add a Local Java Core Module
This contains utilities shared by runners which execute locally. This
is expected to be the DirectRunner and the ReferenceRunner, both of
which can utilize shared in-memory representations about the state of an
executing Pipeline.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/4babcbf4
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/4babcbf4
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/4babcbf4
Branch: refs/heads/master
Commit: 4babcbf46e0a4bada90a77f908eeaa7870138ea9
Parents: 50e58f0
Author: Thomas Groh <tg...@google.com>
Authored: Wed Oct 25 16:31:09 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Nov 17 17:23:41 2017 -0800
----------------------------------------------------------------------
pom.xml | 6 ++
runners/direct-java/pom.xml | 5 ++
.../beam/runners/direct/CommittedBundle.java | 9 ++-
.../direct/ImmutableListBundleFactory.java | 8 +++
.../beam/runners/direct/WatermarkManager.java | 4 +-
.../direct/ImmutableListBundleFactoryTest.java | 4 +-
runners/local-java/pom.xml | 62 ++++++++++++++++++++
.../apache/beam/runners/core/local/Bundle.java | 33 +++++++++++
.../beam/runners/core/local/package-info.java | 22 +++++++
runners/pom.xml | 1 +
10 files changed, 145 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 7bf9d66..c5223df 100644
--- a/pom.xml
+++ b/pom.xml
@@ -702,6 +702,12 @@
<dependency>
<groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-local-java-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.beam</groupId>
<artifactId>beam-runners-direct-java</artifactId>
<version>${project.version}</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/direct-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/direct-java/pom.xml b/runners/direct-java/pom.xml
index 752af44..ff7b4a8 100644
--- a/runners/direct-java/pom.xml
+++ b/runners/direct-java/pom.xml
@@ -182,6 +182,11 @@
<dependency>
<groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-local-java-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
index 79a96fe..ccb031f 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
@@ -20,6 +20,7 @@ package org.apache.beam.runners.direct;
import javax.annotation.Nullable;
import org.apache.beam.runners.core.TimerInternals.TimerData;
+import org.apache.beam.runners.core.local.Bundle;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
@@ -33,7 +34,7 @@ import org.joda.time.Instant;
* a part of at a later point.
* @param <T> the type of elements contained within this bundle
*/
-interface CommittedBundle<T> {
+interface CommittedBundle<T> extends Bundle<T> {
/**
* Returns the PCollection that the elements of this bundle belong to.
*/
@@ -52,10 +53,8 @@ interface CommittedBundle<T> {
*/
Iterable<WindowedValue<T>> getElements();
- /**
- * Returns the minimum timestamp among all of the elements of this {@link CommittedBundle}.
- */
- Instant getMinTimestamp();
+ @Override
+ Instant getMinimumTimestamp();
/**
* Returns the processing time output watermark at the time the producing {@link PTransform}
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
index 73734d0..bea7699 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
@@ -22,6 +22,8 @@ import static com.google.common.base.Preconditions.checkState;
import com.google.auto.value.AutoValue;
import com.google.common.collect.ImmutableList;
+import java.util.Iterator;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.WindowedValue;
@@ -128,6 +130,12 @@ class ImmutableListBundleFactory implements BundleFactory {
}
@Override
+ @Nonnull
+ public Iterator<WindowedValue<T>> iterator() {
+ return getElements().iterator();
+ }
+
+ @Override
public CommittedBundle<T> withElements(Iterable<WindowedValue<T>> elements) {
return create(
getPCollection(),
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
index 599b74f..a1395a9 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
@@ -274,7 +274,7 @@ class WatermarkManager {
if (!pendingElements.isEmpty()) {
minInputWatermark =
INSTANT_ORDERING.min(
- minInputWatermark, pendingElements.firstEntry().getElement().getMinTimestamp());
+ minInputWatermark, pendingElements.firstEntry().getElement().getMinimumTimestamp());
}
Instant newWatermark = INSTANT_ORDERING.max(oldWatermark, minInputWatermark);
currentWatermark.set(newWatermark);
@@ -1511,7 +1511,7 @@ class WatermarkManager {
@Override
public int compare(CommittedBundle<?> o1, CommittedBundle<?> o2) {
return ComparisonChain.start()
- .compare(o1.getMinTimestamp(), o2.getMinTimestamp())
+ .compare(o1.getMinimumTimestamp(), o2.getMinimumTimestamp())
.result();
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
index 4a392db..83426c6 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
@@ -120,7 +120,7 @@ public class ImmutableListBundleFactoryTest {
// Sanity check that the test is meaningful.
assertThat(minElementTs, not(equalTo(commitTime)));
- assertThat(committed.getMinTimestamp(), equalTo(minElementTs));
+ assertThat(committed.getMinimumTimestamp(), equalTo(minElementTs));
assertThat(committed.getSynchronizedProcessingOutputWatermark(), equalTo(commitTime));
return committed;
@@ -190,7 +190,7 @@ public class ImmutableListBundleFactoryTest {
assertThat(
withed.getSynchronizedProcessingOutputWatermark(),
equalTo(committed.getSynchronizedProcessingOutputWatermark()));
- assertThat(withed.getMinTimestamp(), equalTo(new Instant(2048L)));
+ assertThat(withed.getMinimumTimestamp(), equalTo(new Instant(2048L)));
}
@Test
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/local-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/local-java/pom.xml b/runners/local-java/pom.xml
new file mode 100644
index 0000000..b988fd9
--- /dev/null
+++ b/runners/local-java/pom.xml
@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-parent</artifactId>
+ <version>2.3.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>beam-runners-local-java-core</artifactId>
+
+ <name>Apache Beam :: Runners :: Local Java Core</name>
+
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <!-- The Local Java Core artifact should carefully manage the classes made available by this
+ dependency. The Java SDK should be used to provide common utilities (e.g. Coder, WindowedValue)
+ but should not be used within this library to execute any UDFs.
+ TODO: Add an APISurfaceTest to force this to be the case, if possible. -->
+ <dependency>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-sdks-java-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>joda-time</groupId>
+ <artifactId>joda-time</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </dependency>
+
+ <!-- build dependencies -->
+ <dependency>
+ <groupId>com.google.auto.value</groupId>
+ <artifactId>auto-value</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java b/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java
new file mode 100644
index 0000000..98e1e8a
--- /dev/null
+++ b/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java
@@ -0,0 +1,33 @@
+/*
+ * 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.local;
+
+import org.apache.beam.sdk.util.WindowedValue;
+import org.joda.time.Instant;
+
+/** An immutable collection of elements which are part of a {@code PCollection}. */
+public interface Bundle<T> extends Iterable<WindowedValue<T>> {
+ /**
+ * Return the minimum timestamp among elements in this bundle.
+ *
+ * <p>This should be equivalent to iterating over all of the elements within a bundle and
+ * selecting the minimum timestamp from among them.
+ */
+ Instant getMinimumTimestamp();
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java b/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java
new file mode 100644
index 0000000..ffd2e54
--- /dev/null
+++ b/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Utilities useful when executing a pipeline on a single machine.
+ */
+package org.apache.beam.runners.core.local;
http://git-wip-us.apache.org/repos/asf/beam/blob/4babcbf4/runners/pom.xml
----------------------------------------------------------------------
diff --git a/runners/pom.xml b/runners/pom.xml
index 47f3c0e..814b3f1 100644
--- a/runners/pom.xml
+++ b/runners/pom.xml
@@ -36,6 +36,7 @@
<module>core-construction-java</module>
<module>core-java</module>
<module>java-fn-execution</module>
+ <module>local-java</module>
<module>local-artifact-service-java</module>
<module>reference</module>
<module>direct-java</module>
[3/4] beam git commit: Add A TransformExecutorFactory
Posted by tg...@apache.org.
Add A TransformExecutorFactory
This creates executors provided an input bundle, a transform, a
completion callback, and a service to execute the transform on.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/8101103b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/8101103b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/8101103b
Branch: refs/heads/master
Commit: 8101103beb4de82acb1d4785097281c071f124d7
Parents: c8d45d4
Author: Thomas Groh <tg...@google.com>
Authored: Mon Nov 13 18:08:17 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Nov 17 17:23:41 2017 -0800
----------------------------------------------------------------------
.../runners/direct/DirectTransformExecutor.java | 196 +++++++
.../direct/ExecutorServiceParallelExecutor.java | 56 +-
.../beam/runners/direct/TransformExecutor.java | 165 +-----
.../direct/TransformExecutorFactory.java | 32 ++
.../direct/TransformExecutorService.java | 6 +-
.../direct/TransformExecutorServices.java | 14 +-
.../direct/DirectRunnerApiSurfaceTest.java | 5 +-
.../direct/DirectTransformExecutorTest.java | 537 +++++++++++++++++++
.../direct/TransformExecutorServicesTest.java | 40 +-
.../runners/direct/TransformExecutorTest.java | 537 -------------------
10 files changed, 823 insertions(+), 765 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTransformExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTransformExecutor.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTransformExecutor.java
new file mode 100644
index 0000000..a5aa809
--- /dev/null
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTransformExecutor.java
@@ -0,0 +1,196 @@
+/*
+ * 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.direct;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.MoreObjects;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import org.apache.beam.runners.core.construction.PTransformTranslation;
+import org.apache.beam.runners.core.metrics.MetricUpdates;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link Callable} responsible for constructing a {@link TransformEvaluator} from a {@link
+ * TransformEvaluatorFactory} and evaluating it on some bundle of input, and registering the result
+ * using a registered {@link CompletionCallback}.
+ */
+class DirectTransformExecutor<T> implements TransformExecutor {
+ private static final Logger LOG = LoggerFactory.getLogger(DirectTransformExecutor.class);
+
+ static class Factory implements TransformExecutorFactory {
+ private final EvaluationContext context;
+ private final TransformEvaluatorRegistry registry;
+ private final Map<String, Collection<ModelEnforcementFactory>> transformEnforcements;
+
+ Factory(
+ EvaluationContext context,
+ TransformEvaluatorRegistry registry,
+ Map<String, Collection<ModelEnforcementFactory>> transformEnforcements) {
+ this.context = context;
+ this.registry = registry;
+ this.transformEnforcements = transformEnforcements;
+ }
+
+ @Override
+ public TransformExecutor create(
+ CommittedBundle<?> bundle,
+ AppliedPTransform<?, ?, ?> transform,
+ CompletionCallback onComplete,
+ TransformExecutorService executorService) {
+ Collection<ModelEnforcementFactory> enforcements =
+ MoreObjects.firstNonNull(
+ transformEnforcements.get(
+ PTransformTranslation.urnForTransform(transform.getTransform())),
+ Collections.<ModelEnforcementFactory>emptyList());
+ return new DirectTransformExecutor<>(
+ context, registry, enforcements, bundle, transform, onComplete, executorService);
+ }
+ }
+
+ private final TransformEvaluatorFactory evaluatorFactory;
+ private final Iterable<? extends ModelEnforcementFactory> modelEnforcements;
+
+ /** The transform that will be evaluated. */
+ private final AppliedPTransform<?, ?, ?> transform;
+ /** The inputs this {@link DirectTransformExecutor} will deliver to the transform. */
+ private final CommittedBundle<T> inputBundle;
+
+ private final CompletionCallback onComplete;
+ private final TransformExecutorService transformEvaluationState;
+ private final EvaluationContext context;
+
+ @VisibleForTesting
+ DirectTransformExecutor(
+ EvaluationContext context,
+ TransformEvaluatorFactory factory,
+ Iterable<? extends ModelEnforcementFactory> modelEnforcements,
+ CommittedBundle<T> inputBundle,
+ AppliedPTransform<?, ?, ?> transform,
+ CompletionCallback completionCallback,
+ TransformExecutorService transformEvaluationState) {
+ this.evaluatorFactory = factory;
+ this.modelEnforcements = modelEnforcements;
+
+ this.inputBundle = inputBundle;
+ this.transform = transform;
+
+ this.onComplete = completionCallback;
+
+ this.transformEvaluationState = transformEvaluationState;
+ this.context = context;
+ }
+
+ @Override
+ public void run() {
+ MetricsContainerImpl metricsContainer = new MetricsContainerImpl(transform.getFullName());
+ try (Closeable metricsScope = MetricsEnvironment.scopedMetricsContainer(metricsContainer)) {
+ Collection<ModelEnforcement<T>> enforcements = new ArrayList<>();
+ for (ModelEnforcementFactory enforcementFactory : modelEnforcements) {
+ ModelEnforcement<T> enforcement = enforcementFactory.forBundle(inputBundle, transform);
+ enforcements.add(enforcement);
+ }
+ TransformEvaluator<T> evaluator =
+ evaluatorFactory.forApplication(transform, inputBundle);
+ if (evaluator == null) {
+ onComplete.handleEmpty(transform);
+ // Nothing to do
+ return;
+ }
+
+ processElements(evaluator, metricsContainer, enforcements);
+
+ finishBundle(evaluator, metricsContainer, enforcements);
+ } catch (Exception e) {
+ onComplete.handleException(inputBundle, e);
+ if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ }
+ throw new RuntimeException(e);
+ } catch (Error err) {
+ LOG.error("Error occurred within {}", this, err);
+ onComplete.handleError(err);
+ throw err;
+ } finally {
+ // Report the physical metrics from the end of this step.
+ context.getMetrics().commitPhysical(inputBundle, metricsContainer.getCumulative());
+
+ transformEvaluationState.complete(this);
+ }
+ }
+
+ /**
+ * Processes all the elements in the input bundle using the transform evaluator, applying any
+ * necessary {@link ModelEnforcement ModelEnforcements}.
+ */
+ private void processElements(
+ TransformEvaluator<T> evaluator,
+ MetricsContainerImpl metricsContainer,
+ Collection<ModelEnforcement<T>> enforcements)
+ throws Exception {
+ if (inputBundle != null) {
+ for (WindowedValue<T> value : inputBundle.getElements()) {
+ for (ModelEnforcement<T> enforcement : enforcements) {
+ enforcement.beforeElement(value);
+ }
+
+ evaluator.processElement(value);
+
+ // Report the physical metrics after each element
+ MetricUpdates deltas = metricsContainer.getUpdates();
+ if (deltas != null) {
+ context.getMetrics().updatePhysical(inputBundle, deltas);
+ metricsContainer.commitUpdates();
+ }
+
+ for (ModelEnforcement<T> enforcement : enforcements) {
+ enforcement.afterElement(value);
+ }
+ }
+ }
+ }
+
+ /**
+ * Finishes processing the input bundle and commit the result using the
+ * {@link CompletionCallback}, applying any {@link ModelEnforcement} if necessary.
+ *
+ * @return the {@link TransformResult} produced by
+ * {@link TransformEvaluator#finishBundle()}
+ */
+ private TransformResult<T> finishBundle(
+ TransformEvaluator<T> evaluator, MetricsContainerImpl metricsContainer,
+ Collection<ModelEnforcement<T>> enforcements)
+ throws Exception {
+ TransformResult<T> result =
+ evaluator.finishBundle().withLogicalMetricUpdates(metricsContainer.getCumulative());
+ CommittedResult outputs = onComplete.handleResult(inputBundle, result);
+ for (ModelEnforcement<T> enforcement : enforcements) {
+ enforcement.afterFinish(inputBundle, result, outputs.getOutputs());
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
index 75e2562..fe3765b 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
@@ -20,7 +20,6 @@ package org.apache.beam.runners.direct;
import static com.google.common.base.Preconditions.checkState;
import com.google.auto.value.AutoValue;
-import com.google.common.base.MoreObjects;
import com.google.common.base.Optional;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
@@ -49,7 +48,6 @@ import javax.annotation.Nullable;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.core.TimerInternals.TimerData;
-import org.apache.beam.runners.core.construction.PTransformTranslation;
import org.apache.beam.runners.direct.WatermarkManager.FiredTimers;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult.State;
@@ -77,32 +75,33 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor {
private final DirectGraph graph;
private final RootProviderRegistry rootProviderRegistry;
private final TransformEvaluatorRegistry registry;
- private final Map<String, Collection<ModelEnforcementFactory>> transformEnforcements;
private final EvaluationContext evaluationContext;
- private final LoadingCache<StepAndKey, TransformExecutorService> executorServices;
+ private final TransformExecutorFactory executorFactory;
+ private final TransformExecutorService parallelExecutorService;
+ private final LoadingCache<StepAndKey, TransformExecutorService> serialExecutorServices;
private final Queue<ExecutorUpdate> allUpdates;
private final BlockingQueue<VisibleExecutorUpdate> visibleUpdates;
- private final TransformExecutorService parallelExecutorService;
private final CompletionCallback defaultCompletionCallback;
private final ConcurrentMap<AppliedPTransform<?, ?, ?>, ConcurrentLinkedQueue<CommittedBundle<?>>>
pendingRootBundles;
- private final AtomicReference<ExecutorState> state =
+ private final AtomicReference<ExecutorState> state =
new AtomicReference<>(ExecutorState.QUIESCENT);
/**
- * Measures the number of {@link TransformExecutor TransformExecutors} that have been scheduled
- * but not yet completed.
+ * Measures the number of {@link TransformExecutor TransformExecutors} that have been
+ * scheduled but not yet completed.
*
- * <p>Before a {@link TransformExecutor} is scheduled, this value is incremented. All methods in
- * {@link CompletionCallback} decrement this value.
+ * <p>Before a {@link TransformExecutor} is scheduled, this value is incremented. All
+ * methods in {@link CompletionCallback} decrement this value.
*/
private final AtomicLong outstandingWork = new AtomicLong();
+
private AtomicReference<State> pipelineState = new AtomicReference<>(State.RUNNING);
public static ExecutorServiceParallelExecutor create(
@@ -141,13 +140,12 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor {
this.graph = graph;
this.rootProviderRegistry = rootProviderRegistry;
this.registry = registry;
- this.transformEnforcements = transformEnforcements;
this.evaluationContext = context;
// Weak Values allows TransformExecutorServices that are no longer in use to be reclaimed.
// Executing TransformExecutorServices have a strong reference to their TransformExecutorService
// which stops the TransformExecutorServices from being prematurely garbage collected
- executorServices =
+ serialExecutorServices =
CacheBuilder.newBuilder()
.weakValues()
.removalListener(shutdownExecutorServiceListener())
@@ -160,6 +158,7 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor {
defaultCompletionCallback =
new TimerIterableCompletionCallback(Collections.<TimerData>emptyList());
this.pendingRootBundles = new ConcurrentHashMap<>();
+ executorFactory = new DirectTransformExecutor.Factory(context, registry, transformEnforcements);
}
private CacheLoader<StepAndKey, TransformExecutorService>
@@ -222,29 +221,16 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor {
final StepAndKey stepAndKey = StepAndKey.of(transform, bundle.getKey());
// This executor will remain reachable until it has executed all scheduled transforms.
// The TransformExecutors keep a strong reference to the Executor, the ExecutorService keeps
- // a reference to the scheduled TransformExecutor callable. Follow-up TransformExecutors
- // (scheduled due to the completion of another TransformExecutor) are provided to the
- // ExecutorService before the Earlier TransformExecutor callable completes.
- transformExecutor = executorServices.getUnchecked(stepAndKey);
+ // a reference to the scheduled DirectTransformExecutor callable. Follow-up TransformExecutors
+ // (scheduled due to the completion of another DirectTransformExecutor) are provided to the
+ // ExecutorService before the Earlier DirectTransformExecutor callable completes.
+ transformExecutor = serialExecutorServices.getUnchecked(stepAndKey);
} else {
transformExecutor = parallelExecutorService;
}
- Collection<ModelEnforcementFactory> enforcements =
- MoreObjects.firstNonNull(
- transformEnforcements.get(
- PTransformTranslation.urnForTransform(transform.getTransform())),
- Collections.<ModelEnforcementFactory>emptyList());
-
- TransformExecutor<T> callable =
- TransformExecutor.create(
- evaluationContext,
- registry,
- enforcements,
- bundle,
- transform,
- onComplete,
- transformExecutor);
+ TransformExecutor callable =
+ executorFactory.create(bundle, transform, onComplete, transformExecutor);
outstandingWork.incrementAndGet();
if (!pipelineState.get().isTerminal()) {
transformExecutor.schedule(callable);
@@ -321,8 +307,8 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor {
pipelineState.compareAndSet(State.RUNNING, newState);
// Stop accepting new work before shutting down the executor. This ensures that thread don't try
// to add work to the shutdown executor.
- executorServices.invalidateAll();
- executorServices.cleanUp();
+ serialExecutorServices.invalidateAll();
+ serialExecutorServices.cleanUp();
parallelExecutorService.shutdown();
executorService.shutdown();
try {
@@ -584,7 +570,7 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor {
}
/**
- * If all active {@link TransformExecutor TransformExecutors} are in a blocked state,
+ * If all active {@link DirectTransformExecutor TransformExecutors} are in a blocked state,
* add more work from root nodes that may have additional work. This ensures that if a pipeline
* has elements available from the root nodes it will add those elements when necessary.
*/
@@ -621,7 +607,7 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor {
* been evaluated, and all pending, including potentially blocked work, should be evaluated.
*
* <p>The executor becomes active whenever a timer fires, a {@link PCollectionView} is updated,
- * or output is produced by the evaluation of a {@link TransformExecutor}.
+ * or output is produced by the evaluation of a {@link DirectTransformExecutor}.
*/
ACTIVE,
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
index 76d817b..1e269de 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
@@ -15,167 +15,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.direct;
-
-import java.io.Closeable;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.concurrent.Callable;
-import org.apache.beam.runners.core.metrics.MetricUpdates;
-import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
-import org.apache.beam.sdk.metrics.MetricsEnvironment;
-import org.apache.beam.sdk.runners.AppliedPTransform;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A {@link Callable} responsible for constructing a {@link TransformEvaluator} from a
- * {@link TransformEvaluatorFactory} and evaluating it on some bundle of input, and registering
- * the result using a registered {@link CompletionCallback}.
- *
- * <p>A {@link TransformExecutor} that is currently executing also provides access to the thread
- * that it is being executed on.
- */
-class TransformExecutor<T> implements Runnable {
- private static final Logger LOG = LoggerFactory.getLogger(TransformExecutor.class);
-
- public static <T> TransformExecutor<T> create(
- EvaluationContext context,
- TransformEvaluatorFactory factory,
- Iterable<? extends ModelEnforcementFactory> modelEnforcements,
- CommittedBundle<T> inputBundle,
- AppliedPTransform<?, ?, ?> transform,
- CompletionCallback completionCallback,
- TransformExecutorService transformEvaluationState) {
- return new TransformExecutor<>(
- context,
- factory,
- modelEnforcements,
- inputBundle,
- transform,
- completionCallback,
- transformEvaluationState);
- }
-
- private final TransformEvaluatorFactory evaluatorFactory;
- private final Iterable<? extends ModelEnforcementFactory> modelEnforcements;
-
- /** The transform that will be evaluated. */
- private final AppliedPTransform<?, ?, ?> transform;
- /** The inputs this {@link TransformExecutor} will deliver to the transform. */
- private final CommittedBundle<T> inputBundle;
-
- private final CompletionCallback onComplete;
- private final TransformExecutorService transformEvaluationState;
- private final EvaluationContext context;
-
- private TransformExecutor(
- EvaluationContext context,
- TransformEvaluatorFactory factory,
- Iterable<? extends ModelEnforcementFactory> modelEnforcements,
- CommittedBundle<T> inputBundle,
- AppliedPTransform<?, ?, ?> transform,
- CompletionCallback completionCallback,
- TransformExecutorService transformEvaluationState) {
- this.evaluatorFactory = factory;
- this.modelEnforcements = modelEnforcements;
-
- this.inputBundle = inputBundle;
- this.transform = transform;
- this.onComplete = completionCallback;
-
- this.transformEvaluationState = transformEvaluationState;
- this.context = context;
- }
-
- @Override
- public void run() {
- MetricsContainerImpl metricsContainer = new MetricsContainerImpl(transform.getFullName());
- try (Closeable metricsScope = MetricsEnvironment.scopedMetricsContainer(metricsContainer)) {
- Collection<ModelEnforcement<T>> enforcements = new ArrayList<>();
- for (ModelEnforcementFactory enforcementFactory : modelEnforcements) {
- ModelEnforcement<T> enforcement = enforcementFactory.forBundle(inputBundle, transform);
- enforcements.add(enforcement);
- }
- TransformEvaluator<T> evaluator =
- evaluatorFactory.forApplication(transform, inputBundle);
- if (evaluator == null) {
- onComplete.handleEmpty(transform);
- // Nothing to do
- return;
- }
-
- processElements(evaluator, metricsContainer, enforcements);
-
- finishBundle(evaluator, metricsContainer, enforcements);
- } catch (Exception e) {
- onComplete.handleException(inputBundle, e);
- if (e instanceof RuntimeException) {
- throw (RuntimeException) e;
- }
- throw new RuntimeException(e);
- } catch (Error err) {
- LOG.error("Error occurred within {}", this, err);
- onComplete.handleError(err);
- throw err;
- } finally {
- // Report the physical metrics from the end of this step.
- context.getMetrics().commitPhysical(inputBundle, metricsContainer.getCumulative());
-
- transformEvaluationState.complete(this);
- }
- }
-
- /**
- * Processes all the elements in the input bundle using the transform evaluator, applying any
- * necessary {@link ModelEnforcement ModelEnforcements}.
- */
- private void processElements(
- TransformEvaluator<T> evaluator,
- MetricsContainerImpl metricsContainer,
- Collection<ModelEnforcement<T>> enforcements)
- throws Exception {
- if (inputBundle != null) {
- for (WindowedValue<T> value : inputBundle.getElements()) {
- for (ModelEnforcement<T> enforcement : enforcements) {
- enforcement.beforeElement(value);
- }
-
- evaluator.processElement(value);
-
- // Report the physical metrics after each element
- MetricUpdates deltas = metricsContainer.getUpdates();
- if (deltas != null) {
- context.getMetrics().updatePhysical(inputBundle, deltas);
- metricsContainer.commitUpdates();
- }
-
- for (ModelEnforcement<T> enforcement : enforcements) {
- enforcement.afterElement(value);
- }
- }
- }
- }
+package org.apache.beam.runners.direct;
- /**
- * Finishes processing the input bundle and commit the result using the
- * {@link CompletionCallback}, applying any {@link ModelEnforcement} if necessary.
- *
- * @return the {@link TransformResult} produced by
- * {@link TransformEvaluator#finishBundle()}
- */
- private TransformResult<T> finishBundle(
- TransformEvaluator<T> evaluator, MetricsContainerImpl metricsContainer,
- Collection<ModelEnforcement<T>> enforcements)
- throws Exception {
- TransformResult<T> result =
- evaluator.finishBundle().withLogicalMetricUpdates(metricsContainer.getCumulative());
- CommittedResult outputs = onComplete.handleResult(inputBundle, result);
- for (ModelEnforcement<T> enforcement : enforcements) {
- enforcement.afterFinish(inputBundle, result, outputs.getOutputs());
- }
- return result;
- }
-}
+/** A {@link Runnable} that will execute a {@code PTransform} on some bundle of input. */
+public interface TransformExecutor extends Runnable {}
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorFactory.java
new file mode 100644
index 0000000..d78c265
--- /dev/null
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorFactory.java
@@ -0,0 +1,32 @@
+/*
+ * 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.direct;
+
+import org.apache.beam.sdk.runners.AppliedPTransform;
+
+/**
+ * A Factory for creating {@link TransformExecutor Transform Executors} on an input.
+ */
+interface TransformExecutorFactory {
+ TransformExecutor create(
+ CommittedBundle<?> bundle,
+ AppliedPTransform<?, ?, ?> transform,
+ CompletionCallback onComplete,
+ TransformExecutorService executorService);
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorService.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorService.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorService.java
index c6f770f..90b3960 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorService.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorService.java
@@ -25,13 +25,13 @@ interface TransformExecutorService {
/**
* Schedule the provided work to be eventually executed.
*/
- void schedule(TransformExecutor<?> work);
+ void schedule(TransformExecutor work);
/**
* Finish executing the provided work. This may cause additional
- * {@link TransformExecutor TransformExecutors} to be evaluated.
+ * {@link DirectTransformExecutor TransformExecutors} to be evaluated.
*/
- void complete(TransformExecutor<?> completed);
+ void complete(TransformExecutor completed);
/**
* Cancel any outstanding work, if possible. Any future calls to schedule should ignore any
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorServices.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorServices.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorServices.java
index 53087bf..9aa71f7 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorServices.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutorServices.java
@@ -69,7 +69,7 @@ final class TransformExecutorServices {
}
@Override
- public void schedule(TransformExecutor<?> work) {
+ public void schedule(TransformExecutor work) {
if (active.get()) {
try {
executor.submit(work);
@@ -92,7 +92,7 @@ final class TransformExecutorServices {
}
@Override
- public void complete(TransformExecutor<?> completed) {
+ public void complete(TransformExecutor completed) {
}
@Override
@@ -112,8 +112,8 @@ final class TransformExecutorServices {
private static class SerialTransformExecutor implements TransformExecutorService {
private final ExecutorService executor;
- private AtomicReference<TransformExecutor<?>> currentlyEvaluating;
- private final Queue<TransformExecutor<?>> workQueue;
+ private AtomicReference<TransformExecutor> currentlyEvaluating;
+ private final Queue<TransformExecutor> workQueue;
private boolean active = true;
private SerialTransformExecutor(ExecutorService executor) {
@@ -127,13 +127,13 @@ final class TransformExecutorServices {
* evaluated and scheduling it immediately otherwise.
*/
@Override
- public void schedule(TransformExecutor<?> work) {
+ public void schedule(TransformExecutor work) {
workQueue.offer(work);
updateCurrentlyEvaluating();
}
@Override
- public void complete(TransformExecutor<?> completed) {
+ public void complete(TransformExecutor completed) {
if (!currentlyEvaluating.compareAndSet(completed, null)) {
throw new IllegalStateException(
"Finished work "
@@ -156,7 +156,7 @@ final class TransformExecutorServices {
if (currentlyEvaluating.get() == null) {
// Only synchronize if we need to update what's currently evaluating
synchronized (this) {
- TransformExecutor<?> newWork = workQueue.poll();
+ TransformExecutor newWork = workQueue.poll();
if (active && newWork != null) {
if (currentlyEvaluating.compareAndSet(null, newWork)) {
executor.submit(newWork);
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerApiSurfaceTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerApiSurfaceTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerApiSurfaceTest.java
index 631349f..f116709 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerApiSurfaceTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerApiSurfaceTest.java
@@ -42,7 +42,10 @@ public class DirectRunnerApiSurfaceTest {
// The DirectRunner can expose the Core SDK, anything exposed by the Core SDK, and itself
@SuppressWarnings("unchecked")
final Set<String> allowed =
- ImmutableSet.of("org.apache.beam.sdk", "org.apache.beam.runners.direct", "org.joda.time");
+ ImmutableSet.of(
+ "org.apache.beam.sdk",
+ "org.apache.beam.runners.direct",
+ "org.joda.time");
final Package thisPackage = getClass().getPackage();
final ClassLoader thisClassLoader = getClass().getClassLoader();
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTransformExecutorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTransformExecutorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTransformExecutorTest.java
new file mode 100644
index 0000000..aa83e0b
--- /dev/null
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTransformExecutorTest.java
@@ -0,0 +1,537 @@
+/*
+ * 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.direct;
+
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.isA;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.when;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.beam.runners.direct.CommittedResult.OutputType;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.hamcrest.Matchers;
+import org.joda.time.Instant;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/** Tests for {@link DirectTransformExecutor}. */
+@RunWith(JUnit4.class)
+public class DirectTransformExecutorTest {
+ @Rule public ExpectedException thrown = ExpectedException.none();
+ private PCollection<String> created;
+
+ private AppliedPTransform<?, ?, ?> createdProducer;
+ private AppliedPTransform<?, ?, ?> downstreamProducer;
+
+ private CountDownLatch evaluatorCompleted;
+
+ private RegisteringCompletionCallback completionCallback;
+ private TransformExecutorService transformEvaluationState;
+ private BundleFactory bundleFactory;
+ @Mock private DirectMetrics metrics;
+ @Mock private EvaluationContext evaluationContext;
+ @Mock private TransformEvaluatorRegistry registry;
+
+ @Rule
+ public TestPipeline p = TestPipeline.create().enableAbandonedNodeEnforcement(false);
+
+ @Before
+ public void setup() {
+ MockitoAnnotations.initMocks(this);
+
+ bundleFactory = ImmutableListBundleFactory.create();
+
+ transformEvaluationState =
+ TransformExecutorServices.parallel(MoreExecutors.newDirectExecutorService());
+
+ evaluatorCompleted = new CountDownLatch(1);
+ completionCallback = new RegisteringCompletionCallback(evaluatorCompleted);
+
+ created = p.apply(Create.of("foo", "spam", "third"));
+ PCollection<KV<Integer, String>> downstream = created.apply(WithKeys.<Integer, String>of(3));
+
+ DirectGraphs.performDirectOverrides(p);
+ DirectGraph graph = DirectGraphs.getGraph(p);
+ createdProducer = graph.getProducer(created);
+ downstreamProducer = graph.getProducer(downstream);
+
+ when(evaluationContext.getMetrics()).thenReturn(metrics);
+ }
+
+ @Test
+ public void callWithNullInputBundleFinishesBundleAndCompletes() throws Exception {
+ final TransformResult<Object> result = StepTransformResult.withoutHold(createdProducer).build();
+ final AtomicBoolean finishCalled = new AtomicBoolean(false);
+ TransformEvaluator<Object> evaluator =
+ new TransformEvaluator<Object>() {
+ @Override
+ public void processElement(WindowedValue<Object> element) throws Exception {
+ throw new IllegalArgumentException("Shouldn't be called");
+ }
+
+ @Override
+ public TransformResult<Object> finishBundle() throws Exception {
+ finishCalled.set(true);
+ return result;
+ }
+ };
+
+ when(registry.forApplication(createdProducer, null)).thenReturn(evaluator);
+
+ DirectTransformExecutor<Object> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>emptyList(),
+ null,
+ createdProducer,
+ completionCallback,
+ transformEvaluationState);
+ executor.run();
+
+ assertThat(finishCalled.get(), is(true));
+ assertThat(completionCallback.handledResult, Matchers.<TransformResult<?>>equalTo(result));
+ assertThat(completionCallback.handledException, is(nullValue()));
+ }
+
+ @Test
+ public void nullTransformEvaluatorTerminates() throws Exception {
+ when(registry.forApplication(createdProducer, null)).thenReturn(null);
+
+ DirectTransformExecutor<Object> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>emptyList(),
+ null,
+ createdProducer,
+ completionCallback,
+ transformEvaluationState);
+ executor.run();
+
+ assertThat(completionCallback.handledResult, is(nullValue()));
+ assertThat(completionCallback.handledEmpty, equalTo(true));
+ assertThat(completionCallback.handledException, is(nullValue()));
+ }
+
+ @Test
+ public void inputBundleProcessesEachElementFinishesAndCompletes() throws Exception {
+ final TransformResult<String> result =
+ StepTransformResult.<String>withoutHold(downstreamProducer).build();
+ final Collection<WindowedValue<String>> elementsProcessed = new ArrayList<>();
+ TransformEvaluator<String> evaluator =
+ new TransformEvaluator<String>() {
+ @Override
+ public void processElement(WindowedValue<String> element) throws Exception {
+ elementsProcessed.add(element);
+ return;
+ }
+
+ @Override
+ public TransformResult<String> finishBundle() throws Exception {
+ return result;
+ }
+ };
+
+ WindowedValue<String> foo = WindowedValue.valueInGlobalWindow("foo");
+ WindowedValue<String> spam = WindowedValue.valueInGlobalWindow("spam");
+ WindowedValue<String> third = WindowedValue.valueInGlobalWindow("third");
+ CommittedBundle<String> inputBundle =
+ bundleFactory.createBundle(created).add(foo).add(spam).add(third).commit(Instant.now());
+ when(registry.<String>forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
+
+ DirectTransformExecutor<String> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>emptyList(),
+ inputBundle,
+ downstreamProducer,
+ completionCallback,
+ transformEvaluationState);
+
+ Executors.newSingleThreadExecutor().submit(executor);
+
+ evaluatorCompleted.await();
+
+ assertThat(elementsProcessed, containsInAnyOrder(spam, third, foo));
+ assertThat(completionCallback.handledResult, Matchers.<TransformResult<?>>equalTo(result));
+ assertThat(completionCallback.handledException, is(nullValue()));
+ }
+
+ @Test
+ public void processElementThrowsExceptionCallsback() throws Exception {
+ final TransformResult<String> result =
+ StepTransformResult.<String>withoutHold(downstreamProducer).build();
+ final Exception exception = new Exception();
+ TransformEvaluator<String> evaluator =
+ new TransformEvaluator<String>() {
+ @Override
+ public void processElement(WindowedValue<String> element) throws Exception {
+ throw exception;
+ }
+
+ @Override
+ public TransformResult<String> finishBundle() throws Exception {
+ return result;
+ }
+ };
+
+ WindowedValue<String> foo = WindowedValue.valueInGlobalWindow("foo");
+ CommittedBundle<String> inputBundle =
+ bundleFactory.createBundle(created).add(foo).commit(Instant.now());
+ when(registry.<String>forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
+
+ DirectTransformExecutor<String> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>emptyList(),
+ inputBundle,
+ downstreamProducer,
+ completionCallback,
+ transformEvaluationState);
+ Executors.newSingleThreadExecutor().submit(executor);
+
+ evaluatorCompleted.await();
+
+ assertThat(completionCallback.handledResult, is(nullValue()));
+ assertThat(completionCallback.handledException, Matchers.<Throwable>equalTo(exception));
+ }
+
+ @Test
+ public void finishBundleThrowsExceptionCallsback() throws Exception {
+ final Exception exception = new Exception();
+ TransformEvaluator<String> evaluator =
+ new TransformEvaluator<String>() {
+ @Override
+ public void processElement(WindowedValue<String> element) throws Exception {}
+
+ @Override
+ public TransformResult<String> finishBundle() throws Exception {
+ throw exception;
+ }
+ };
+
+ CommittedBundle<String> inputBundle = bundleFactory.createBundle(created).commit(Instant.now());
+ when(registry.<String>forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
+
+ DirectTransformExecutor<String> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>emptyList(),
+ inputBundle,
+ downstreamProducer,
+ completionCallback,
+ transformEvaluationState);
+ Executors.newSingleThreadExecutor().submit(executor);
+
+ evaluatorCompleted.await();
+
+ assertThat(completionCallback.handledResult, is(nullValue()));
+ assertThat(completionCallback.handledException, Matchers.<Throwable>equalTo(exception));
+ }
+
+ @Test
+ public void callWithEnforcementAppliesEnforcement() throws Exception {
+ final TransformResult<Object> result =
+ StepTransformResult.withoutHold(downstreamProducer).build();
+
+ TransformEvaluator<Object> evaluator =
+ new TransformEvaluator<Object>() {
+ @Override
+ public void processElement(WindowedValue<Object> element) throws Exception {}
+
+ @Override
+ public TransformResult<Object> finishBundle() throws Exception {
+ return result;
+ }
+ };
+
+ WindowedValue<String> fooElem = WindowedValue.valueInGlobalWindow("foo");
+ WindowedValue<String> barElem = WindowedValue.valueInGlobalWindow("bar");
+ CommittedBundle<String> inputBundle =
+ bundleFactory.createBundle(created).add(fooElem).add(barElem).commit(Instant.now());
+ when(registry.forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
+
+ TestEnforcementFactory enforcement = new TestEnforcementFactory();
+ DirectTransformExecutor<String> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>singleton(enforcement),
+ inputBundle,
+ downstreamProducer,
+ completionCallback,
+ transformEvaluationState);
+
+ executor.run();
+ TestEnforcement<?> testEnforcement = enforcement.instance;
+ assertThat(
+ testEnforcement.beforeElements,
+ Matchers.<WindowedValue<?>>containsInAnyOrder(barElem, fooElem));
+ assertThat(
+ testEnforcement.afterElements,
+ Matchers.<WindowedValue<?>>containsInAnyOrder(barElem, fooElem));
+ assertThat(testEnforcement.finishedBundles, Matchers.<TransformResult<?>>contains(result));
+ }
+
+ @Test
+ public void callWithEnforcementThrowsOnFinishPropagates() throws Exception {
+ final TransformResult<Object> result =
+ StepTransformResult.withoutHold(createdProducer).build();
+
+ TransformEvaluator<Object> evaluator =
+ new TransformEvaluator<Object>() {
+ @Override
+ public void processElement(WindowedValue<Object> element) throws Exception {}
+
+ @Override
+ public TransformResult<Object> finishBundle() throws Exception {
+ return result;
+ }
+ };
+
+ WindowedValue<String> fooBytes = WindowedValue.valueInGlobalWindow("foo");
+ CommittedBundle<String> inputBundle =
+ bundleFactory.createBundle(created).add(fooBytes).commit(Instant.now());
+ when(registry.forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
+
+ DirectTransformExecutor<String> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>singleton(
+ new ThrowingEnforcementFactory(ThrowingEnforcementFactory.When.AFTER_BUNDLE)),
+ inputBundle,
+ downstreamProducer,
+ completionCallback,
+ transformEvaluationState);
+
+ Future<?> task = Executors.newSingleThreadExecutor().submit(executor);
+
+ thrown.expectCause(isA(RuntimeException.class));
+ thrown.expectMessage("afterFinish");
+ task.get();
+ }
+
+ @Test
+ public void callWithEnforcementThrowsOnElementPropagates() throws Exception {
+ final TransformResult<Object> result =
+ StepTransformResult.withoutHold(createdProducer).build();
+
+ TransformEvaluator<Object> evaluator =
+ new TransformEvaluator<Object>() {
+ @Override
+ public void processElement(WindowedValue<Object> element) throws Exception {}
+
+ @Override
+ public TransformResult<Object> finishBundle() throws Exception {
+ return result;
+ }
+ };
+
+ WindowedValue<String> fooBytes = WindowedValue.valueInGlobalWindow("foo");
+ CommittedBundle<String> inputBundle =
+ bundleFactory.createBundle(created).add(fooBytes).commit(Instant.now());
+ when(registry.forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
+
+ DirectTransformExecutor<String> executor =
+ new DirectTransformExecutor<>(
+ evaluationContext,
+ registry,
+ Collections.<ModelEnforcementFactory>singleton(
+ new ThrowingEnforcementFactory(ThrowingEnforcementFactory.When.AFTER_ELEMENT)),
+ inputBundle,
+ downstreamProducer,
+ completionCallback,
+ transformEvaluationState);
+
+ Future<?> task = Executors.newSingleThreadExecutor().submit(executor);
+
+ thrown.expectCause(isA(RuntimeException.class));
+ thrown.expectMessage("afterElement");
+ task.get();
+ }
+
+ private static class RegisteringCompletionCallback implements CompletionCallback {
+ private TransformResult<?> handledResult = null;
+ private boolean handledEmpty = false;
+ private Exception handledException = null;
+ private final CountDownLatch onMethod;
+
+ private RegisteringCompletionCallback(CountDownLatch onMethod) {
+ this.onMethod = onMethod;
+ }
+
+ @Override
+ public CommittedResult handleResult(CommittedBundle<?> inputBundle, TransformResult<?> result) {
+ handledResult = result;
+ onMethod.countDown();
+ @SuppressWarnings("rawtypes")
+ Iterable unprocessedElements =
+ result.getUnprocessedElements() == null
+ ? Collections.emptyList()
+ : result.getUnprocessedElements();
+
+ Optional<? extends CommittedBundle<?>> unprocessedBundle;
+ if (inputBundle == null || Iterables.isEmpty(unprocessedElements)) {
+ unprocessedBundle = Optional.absent();
+ } else {
+ unprocessedBundle =
+ Optional.<CommittedBundle<?>>of(inputBundle.withElements(unprocessedElements));
+ }
+ return CommittedResult.create(
+ result,
+ unprocessedBundle,
+ Collections.<CommittedBundle<?>>emptyList(),
+ EnumSet.noneOf(OutputType.class));
+ }
+
+ @Override
+ public void handleEmpty(AppliedPTransform<?, ?, ?> transform) {
+ handledEmpty = true;
+ onMethod.countDown();
+ }
+
+ @Override
+ public void handleException(CommittedBundle<?> inputBundle, Exception e) {
+ handledException = e;
+ onMethod.countDown();
+ }
+
+ @Override
+ public void handleError(Error err) {
+ throw err;
+ }
+ }
+
+ private static class TestEnforcementFactory implements ModelEnforcementFactory {
+ private TestEnforcement<?> instance;
+
+ @Override
+ public <T> TestEnforcement<T> forBundle(
+ CommittedBundle<T> input, AppliedPTransform<?, ?, ?> consumer) {
+ TestEnforcement<T> newEnforcement = new TestEnforcement<>();
+ instance = newEnforcement;
+ return newEnforcement;
+ }
+ }
+
+ private static class TestEnforcement<T> implements ModelEnforcement<T> {
+ private final List<WindowedValue<T>> beforeElements = new ArrayList<>();
+ private final List<WindowedValue<T>> afterElements = new ArrayList<>();
+ private final List<TransformResult<?>> finishedBundles = new ArrayList<>();
+
+ @Override
+ public void beforeElement(WindowedValue<T> element) {
+ beforeElements.add(element);
+ }
+
+ @Override
+ public void afterElement(WindowedValue<T> element) {
+ afterElements.add(element);
+ }
+
+ @Override
+ public void afterFinish(
+ CommittedBundle<T> input,
+ TransformResult<T> result,
+ Iterable<? extends CommittedBundle<?>> outputs) {
+ finishedBundles.add(result);
+ }
+ }
+
+ private static class ThrowingEnforcementFactory implements ModelEnforcementFactory {
+ private final When when;
+
+ private ThrowingEnforcementFactory(When when) {
+ this.when = when;
+ }
+
+ enum When {
+ BEFORE_BUNDLE,
+ BEFORE_ELEMENT,
+ AFTER_ELEMENT,
+ AFTER_BUNDLE
+ }
+
+ @Override
+ public <T> ModelEnforcement<T> forBundle(
+ CommittedBundle<T> input, AppliedPTransform<?, ?, ?> consumer) {
+ if (when == When.BEFORE_BUNDLE) {
+ throw new RuntimeException("forBundle");
+ }
+ return new ThrowingEnforcement<>();
+ }
+
+ private class ThrowingEnforcement<T> implements ModelEnforcement<T> {
+ @Override
+ public void beforeElement(WindowedValue<T> element) {
+ if (when == When.BEFORE_ELEMENT) {
+ throw new RuntimeException("beforeElement");
+ }
+ }
+
+ @Override
+ public void afterElement(WindowedValue<T> element) {
+ if (when == When.AFTER_ELEMENT) {
+ throw new RuntimeException("afterElement");
+ }
+ }
+
+ @Override
+ public void afterFinish(
+ CommittedBundle<T> input,
+ TransformResult<T> result,
+ Iterable<? extends CommittedBundle<?>> outputs) {
+ if (when == When.AFTER_BUNDLE) {
+ throw new RuntimeException("afterFinish");
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorServicesTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorServicesTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorServicesTest.java
index 77652b2..5d1c994 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorServicesTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorServicesTest.java
@@ -46,10 +46,10 @@ public class TransformExecutorServicesTest {
@Test
public void parallelScheduleMultipleSchedulesBothImmediately() {
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> first = mock(TransformExecutor.class);
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> second = mock(TransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ first = mock(DirectTransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ second = mock(DirectTransformExecutor.class);
TransformExecutorService parallel =
TransformExecutorServices.parallel(executorService);
@@ -65,8 +65,8 @@ public class TransformExecutorServicesTest {
@Test
public void parallelRejectedStillActiveThrows() {
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> first = mock(TransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ first = mock(DirectTransformExecutor.class);
TransformExecutorService parallel =
TransformExecutorServices.parallel(executorService);
@@ -78,8 +78,8 @@ public class TransformExecutorServicesTest {
@Test
public void parallelRejectedShutdownSucceeds() {
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> first = mock(TransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ first = mock(DirectTransformExecutor.class);
TransformExecutorService parallel =
TransformExecutorServices.parallel(executorService);
@@ -90,10 +90,10 @@ public class TransformExecutorServicesTest {
@Test
public void serialScheduleTwoWaitsForFirstToComplete() {
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> first = mock(TransformExecutor.class);
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> second = mock(TransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ first = mock(DirectTransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ second = mock(DirectTransformExecutor.class);
TransformExecutorService serial = TransformExecutorServices.serial(executorService);
serial.schedule(first);
@@ -110,10 +110,10 @@ public class TransformExecutorServicesTest {
@Test
public void serialCompleteNotExecutingTaskThrows() {
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> first = mock(TransformExecutor.class);
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> second = mock(TransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ first = mock(DirectTransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ second = mock(DirectTransformExecutor.class);
TransformExecutorService serial = TransformExecutorServices.serial(executorService);
serial.schedule(first);
@@ -129,10 +129,10 @@ public class TransformExecutorServicesTest {
*/
@Test
public void serialShutdownCompleteActive() {
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> first = mock(TransformExecutor.class);
- @SuppressWarnings("unchecked")
- TransformExecutor<Object> second = mock(TransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ first = mock(DirectTransformExecutor.class);
+ @SuppressWarnings("unchecked") DirectTransformExecutor<Object>
+ second = mock(DirectTransformExecutor.class);
TransformExecutorService serial = TransformExecutorServices.serial(executorService);
serial.schedule(first);
http://git-wip-us.apache.org/repos/asf/beam/blob/8101103b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
deleted file mode 100644
index b7f5a7c..0000000
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
+++ /dev/null
@@ -1,537 +0,0 @@
-/*
- * 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.direct;
-
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.isA;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.Assert.assertThat;
-import static org.mockito.Mockito.when;
-
-import com.google.common.base.Optional;
-import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.MoreExecutors;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.EnumSet;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.beam.runners.direct.CommittedResult.OutputType;
-import org.apache.beam.sdk.runners.AppliedPTransform;
-import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.WithKeys;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.values.KV;
-import org.apache.beam.sdk.values.PCollection;
-import org.hamcrest.Matchers;
-import org.joda.time.Instant;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-/** Tests for {@link TransformExecutor}. */
-@RunWith(JUnit4.class)
-public class TransformExecutorTest {
- @Rule public ExpectedException thrown = ExpectedException.none();
- private PCollection<String> created;
-
- private AppliedPTransform<?, ?, ?> createdProducer;
- private AppliedPTransform<?, ?, ?> downstreamProducer;
-
- private CountDownLatch evaluatorCompleted;
-
- private RegisteringCompletionCallback completionCallback;
- private TransformExecutorService transformEvaluationState;
- private BundleFactory bundleFactory;
- @Mock private DirectMetrics metrics;
- @Mock private EvaluationContext evaluationContext;
- @Mock private TransformEvaluatorRegistry registry;
-
- @Rule
- public TestPipeline p = TestPipeline.create().enableAbandonedNodeEnforcement(false);
-
- @Before
- public void setup() {
- MockitoAnnotations.initMocks(this);
-
- bundleFactory = ImmutableListBundleFactory.create();
-
- transformEvaluationState =
- TransformExecutorServices.parallel(MoreExecutors.newDirectExecutorService());
-
- evaluatorCompleted = new CountDownLatch(1);
- completionCallback = new RegisteringCompletionCallback(evaluatorCompleted);
-
- created = p.apply(Create.of("foo", "spam", "third"));
- PCollection<KV<Integer, String>> downstream = created.apply(WithKeys.<Integer, String>of(3));
-
- DirectGraphs.performDirectOverrides(p);
- DirectGraph graph = DirectGraphs.getGraph(p);
- createdProducer = graph.getProducer(created);
- downstreamProducer = graph.getProducer(downstream);
-
- when(evaluationContext.getMetrics()).thenReturn(metrics);
- }
-
- @Test
- public void callWithNullInputBundleFinishesBundleAndCompletes() throws Exception {
- final TransformResult<Object> result = StepTransformResult.withoutHold(createdProducer).build();
- final AtomicBoolean finishCalled = new AtomicBoolean(false);
- TransformEvaluator<Object> evaluator =
- new TransformEvaluator<Object>() {
- @Override
- public void processElement(WindowedValue<Object> element) throws Exception {
- throw new IllegalArgumentException("Shouldn't be called");
- }
-
- @Override
- public TransformResult<Object> finishBundle() throws Exception {
- finishCalled.set(true);
- return result;
- }
- };
-
- when(registry.forApplication(createdProducer, null)).thenReturn(evaluator);
-
- TransformExecutor<Object> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>emptyList(),
- null,
- createdProducer,
- completionCallback,
- transformEvaluationState);
- executor.run();
-
- assertThat(finishCalled.get(), is(true));
- assertThat(completionCallback.handledResult, Matchers.<TransformResult<?>>equalTo(result));
- assertThat(completionCallback.handledException, is(nullValue()));
- }
-
- @Test
- public void nullTransformEvaluatorTerminates() throws Exception {
- when(registry.forApplication(createdProducer, null)).thenReturn(null);
-
- TransformExecutor<Object> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>emptyList(),
- null,
- createdProducer,
- completionCallback,
- transformEvaluationState);
- executor.run();
-
- assertThat(completionCallback.handledResult, is(nullValue()));
- assertThat(completionCallback.handledEmpty, equalTo(true));
- assertThat(completionCallback.handledException, is(nullValue()));
- }
-
- @Test
- public void inputBundleProcessesEachElementFinishesAndCompletes() throws Exception {
- final TransformResult<String> result =
- StepTransformResult.<String>withoutHold(downstreamProducer).build();
- final Collection<WindowedValue<String>> elementsProcessed = new ArrayList<>();
- TransformEvaluator<String> evaluator =
- new TransformEvaluator<String>() {
- @Override
- public void processElement(WindowedValue<String> element) throws Exception {
- elementsProcessed.add(element);
- return;
- }
-
- @Override
- public TransformResult<String> finishBundle() throws Exception {
- return result;
- }
- };
-
- WindowedValue<String> foo = WindowedValue.valueInGlobalWindow("foo");
- WindowedValue<String> spam = WindowedValue.valueInGlobalWindow("spam");
- WindowedValue<String> third = WindowedValue.valueInGlobalWindow("third");
- CommittedBundle<String> inputBundle =
- bundleFactory.createBundle(created).add(foo).add(spam).add(third).commit(Instant.now());
- when(registry.<String>forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
-
- TransformExecutor<String> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>emptyList(),
- inputBundle,
- downstreamProducer,
- completionCallback,
- transformEvaluationState);
-
- Executors.newSingleThreadExecutor().submit(executor);
-
- evaluatorCompleted.await();
-
- assertThat(elementsProcessed, containsInAnyOrder(spam, third, foo));
- assertThat(completionCallback.handledResult, Matchers.<TransformResult<?>>equalTo(result));
- assertThat(completionCallback.handledException, is(nullValue()));
- }
-
- @Test
- public void processElementThrowsExceptionCallsback() throws Exception {
- final TransformResult<String> result =
- StepTransformResult.<String>withoutHold(downstreamProducer).build();
- final Exception exception = new Exception();
- TransformEvaluator<String> evaluator =
- new TransformEvaluator<String>() {
- @Override
- public void processElement(WindowedValue<String> element) throws Exception {
- throw exception;
- }
-
- @Override
- public TransformResult<String> finishBundle() throws Exception {
- return result;
- }
- };
-
- WindowedValue<String> foo = WindowedValue.valueInGlobalWindow("foo");
- CommittedBundle<String> inputBundle =
- bundleFactory.createBundle(created).add(foo).commit(Instant.now());
- when(registry.<String>forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
-
- TransformExecutor<String> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>emptyList(),
- inputBundle,
- downstreamProducer,
- completionCallback,
- transformEvaluationState);
- Executors.newSingleThreadExecutor().submit(executor);
-
- evaluatorCompleted.await();
-
- assertThat(completionCallback.handledResult, is(nullValue()));
- assertThat(completionCallback.handledException, Matchers.<Throwable>equalTo(exception));
- }
-
- @Test
- public void finishBundleThrowsExceptionCallsback() throws Exception {
- final Exception exception = new Exception();
- TransformEvaluator<String> evaluator =
- new TransformEvaluator<String>() {
- @Override
- public void processElement(WindowedValue<String> element) throws Exception {}
-
- @Override
- public TransformResult<String> finishBundle() throws Exception {
- throw exception;
- }
- };
-
- CommittedBundle<String> inputBundle = bundleFactory.createBundle(created).commit(Instant.now());
- when(registry.<String>forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
-
- TransformExecutor<String> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>emptyList(),
- inputBundle,
- downstreamProducer,
- completionCallback,
- transformEvaluationState);
- Executors.newSingleThreadExecutor().submit(executor);
-
- evaluatorCompleted.await();
-
- assertThat(completionCallback.handledResult, is(nullValue()));
- assertThat(completionCallback.handledException, Matchers.<Throwable>equalTo(exception));
- }
-
- @Test
- public void callWithEnforcementAppliesEnforcement() throws Exception {
- final TransformResult<Object> result =
- StepTransformResult.withoutHold(downstreamProducer).build();
-
- TransformEvaluator<Object> evaluator =
- new TransformEvaluator<Object>() {
- @Override
- public void processElement(WindowedValue<Object> element) throws Exception {}
-
- @Override
- public TransformResult<Object> finishBundle() throws Exception {
- return result;
- }
- };
-
- WindowedValue<String> fooElem = WindowedValue.valueInGlobalWindow("foo");
- WindowedValue<String> barElem = WindowedValue.valueInGlobalWindow("bar");
- CommittedBundle<String> inputBundle =
- bundleFactory.createBundle(created).add(fooElem).add(barElem).commit(Instant.now());
- when(registry.forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
-
- TestEnforcementFactory enforcement = new TestEnforcementFactory();
- TransformExecutor<String> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>singleton(enforcement),
- inputBundle,
- downstreamProducer,
- completionCallback,
- transformEvaluationState);
-
- executor.run();
- TestEnforcement<?> testEnforcement = enforcement.instance;
- assertThat(
- testEnforcement.beforeElements,
- Matchers.<WindowedValue<?>>containsInAnyOrder(barElem, fooElem));
- assertThat(
- testEnforcement.afterElements,
- Matchers.<WindowedValue<?>>containsInAnyOrder(barElem, fooElem));
- assertThat(testEnforcement.finishedBundles, Matchers.<TransformResult<?>>contains(result));
- }
-
- @Test
- public void callWithEnforcementThrowsOnFinishPropagates() throws Exception {
- final TransformResult<Object> result =
- StepTransformResult.withoutHold(createdProducer).build();
-
- TransformEvaluator<Object> evaluator =
- new TransformEvaluator<Object>() {
- @Override
- public void processElement(WindowedValue<Object> element) throws Exception {}
-
- @Override
- public TransformResult<Object> finishBundle() throws Exception {
- return result;
- }
- };
-
- WindowedValue<String> fooBytes = WindowedValue.valueInGlobalWindow("foo");
- CommittedBundle<String> inputBundle =
- bundleFactory.createBundle(created).add(fooBytes).commit(Instant.now());
- when(registry.forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
-
- TransformExecutor<String> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>singleton(
- new ThrowingEnforcementFactory(ThrowingEnforcementFactory.When.AFTER_BUNDLE)),
- inputBundle,
- downstreamProducer,
- completionCallback,
- transformEvaluationState);
-
- Future<?> task = Executors.newSingleThreadExecutor().submit(executor);
-
- thrown.expectCause(isA(RuntimeException.class));
- thrown.expectMessage("afterFinish");
- task.get();
- }
-
- @Test
- public void callWithEnforcementThrowsOnElementPropagates() throws Exception {
- final TransformResult<Object> result =
- StepTransformResult.withoutHold(createdProducer).build();
-
- TransformEvaluator<Object> evaluator =
- new TransformEvaluator<Object>() {
- @Override
- public void processElement(WindowedValue<Object> element) throws Exception {}
-
- @Override
- public TransformResult<Object> finishBundle() throws Exception {
- return result;
- }
- };
-
- WindowedValue<String> fooBytes = WindowedValue.valueInGlobalWindow("foo");
- CommittedBundle<String> inputBundle =
- bundleFactory.createBundle(created).add(fooBytes).commit(Instant.now());
- when(registry.forApplication(downstreamProducer, inputBundle)).thenReturn(evaluator);
-
- TransformExecutor<String> executor =
- TransformExecutor.create(
- evaluationContext,
- registry,
- Collections.<ModelEnforcementFactory>singleton(
- new ThrowingEnforcementFactory(ThrowingEnforcementFactory.When.AFTER_ELEMENT)),
- inputBundle,
- downstreamProducer,
- completionCallback,
- transformEvaluationState);
-
- Future<?> task = Executors.newSingleThreadExecutor().submit(executor);
-
- thrown.expectCause(isA(RuntimeException.class));
- thrown.expectMessage("afterElement");
- task.get();
- }
-
- private static class RegisteringCompletionCallback implements CompletionCallback {
- private TransformResult<?> handledResult = null;
- private boolean handledEmpty = false;
- private Exception handledException = null;
- private final CountDownLatch onMethod;
-
- private RegisteringCompletionCallback(CountDownLatch onMethod) {
- this.onMethod = onMethod;
- }
-
- @Override
- public CommittedResult handleResult(CommittedBundle<?> inputBundle, TransformResult<?> result) {
- handledResult = result;
- onMethod.countDown();
- @SuppressWarnings("rawtypes")
- Iterable unprocessedElements =
- result.getUnprocessedElements() == null
- ? Collections.emptyList()
- : result.getUnprocessedElements();
-
- Optional<? extends CommittedBundle<?>> unprocessedBundle;
- if (inputBundle == null || Iterables.isEmpty(unprocessedElements)) {
- unprocessedBundle = Optional.absent();
- } else {
- unprocessedBundle =
- Optional.<CommittedBundle<?>>of(inputBundle.withElements(unprocessedElements));
- }
- return CommittedResult.create(
- result,
- unprocessedBundle,
- Collections.<CommittedBundle<?>>emptyList(),
- EnumSet.noneOf(OutputType.class));
- }
-
- @Override
- public void handleEmpty(AppliedPTransform<?, ?, ?> transform) {
- handledEmpty = true;
- onMethod.countDown();
- }
-
- @Override
- public void handleException(CommittedBundle<?> inputBundle, Exception e) {
- handledException = e;
- onMethod.countDown();
- }
-
- @Override
- public void handleError(Error err) {
- throw err;
- }
- }
-
- private static class TestEnforcementFactory implements ModelEnforcementFactory {
- private TestEnforcement<?> instance;
-
- @Override
- public <T> TestEnforcement<T> forBundle(
- CommittedBundle<T> input, AppliedPTransform<?, ?, ?> consumer) {
- TestEnforcement<T> newEnforcement = new TestEnforcement<>();
- instance = newEnforcement;
- return newEnforcement;
- }
- }
-
- private static class TestEnforcement<T> implements ModelEnforcement<T> {
- private final List<WindowedValue<T>> beforeElements = new ArrayList<>();
- private final List<WindowedValue<T>> afterElements = new ArrayList<>();
- private final List<TransformResult<?>> finishedBundles = new ArrayList<>();
-
- @Override
- public void beforeElement(WindowedValue<T> element) {
- beforeElements.add(element);
- }
-
- @Override
- public void afterElement(WindowedValue<T> element) {
- afterElements.add(element);
- }
-
- @Override
- public void afterFinish(
- CommittedBundle<T> input,
- TransformResult<T> result,
- Iterable<? extends CommittedBundle<?>> outputs) {
- finishedBundles.add(result);
- }
- }
-
- private static class ThrowingEnforcementFactory implements ModelEnforcementFactory {
- private final When when;
-
- private ThrowingEnforcementFactory(When when) {
- this.when = when;
- }
-
- enum When {
- BEFORE_BUNDLE,
- BEFORE_ELEMENT,
- AFTER_ELEMENT,
- AFTER_BUNDLE
- }
-
- @Override
- public <T> ModelEnforcement<T> forBundle(
- CommittedBundle<T> input, AppliedPTransform<?, ?, ?> consumer) {
- if (when == When.BEFORE_BUNDLE) {
- throw new RuntimeException("forBundle");
- }
- return new ThrowingEnforcement<>();
- }
-
- private class ThrowingEnforcement<T> implements ModelEnforcement<T> {
- @Override
- public void beforeElement(WindowedValue<T> element) {
- if (when == When.BEFORE_ELEMENT) {
- throw new RuntimeException("beforeElement");
- }
- }
-
- @Override
- public void afterElement(WindowedValue<T> element) {
- if (when == When.AFTER_ELEMENT) {
- throw new RuntimeException("afterElement");
- }
- }
-
- @Override
- public void afterFinish(
- CommittedBundle<T> input,
- TransformResult<T> result,
- Iterable<? extends CommittedBundle<?>> outputs) {
- if (when == When.AFTER_BUNDLE) {
- throw new RuntimeException("afterFinish");
- }
- }
- }
- }
-}
[2/4] beam git commit: Move StructuralKey to the local execution
module
Posted by tg...@apache.org.
Move StructuralKey to the local execution module
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/c8d45d4a
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/c8d45d4a
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/c8d45d4a
Branch: refs/heads/master
Commit: c8d45d4afd39bdc42f90720637bdabd2084aaa30
Parents: 4babcbf
Author: Thomas Groh <tg...@google.com>
Authored: Wed Nov 8 18:39:21 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Nov 17 17:23:41 2017 -0800
----------------------------------------------------------------------
runners/direct-java/pom.xml | 1 +
.../beam/runners/direct/BundleFactory.java | 1 +
.../runners/direct/CloningBundleFactory.java | 1 +
.../beam/runners/direct/CommittedBundle.java | 3 +-
.../runners/direct/DirectExecutionContext.java | 1 +
.../beam/runners/direct/EvaluationContext.java | 1 +
.../GroupAlsoByWindowEvaluatorFactory.java | 1 +
.../direct/GroupByKeyOnlyEvaluatorFactory.java | 1 +
.../ImmutabilityCheckingBundleFactory.java | 1 +
.../direct/ImmutableListBundleFactory.java | 1 +
.../beam/runners/direct/MultiStepCombine.java | 1 +
.../beam/runners/direct/ParDoEvaluator.java | 1 +
.../runners/direct/ParDoEvaluatorFactory.java | 1 +
.../direct/StatefulParDoEvaluatorFactory.java | 1 +
.../apache/beam/runners/direct/StepAndKey.java | 1 +
.../beam/runners/direct/StructuralKey.java | 103 -------------------
.../direct/UnboundedReadDeduplicator.java | 1 +
.../beam/runners/direct/WatermarkManager.java | 1 +
.../direct/CloningBundleFactoryTest.java | 1 +
.../direct/DirectTimerInternalsTest.java | 1 +
.../runners/direct/EvaluationContextTest.java | 1 +
.../direct/GroupByKeyEvaluatorFactoryTest.java | 1 +
.../GroupByKeyOnlyEvaluatorFactoryTest.java | 1 +
.../ImmutabilityCheckingBundleFactoryTest.java | 1 +
.../direct/ImmutableListBundleFactoryTest.java | 1 +
.../beam/runners/direct/ParDoEvaluatorTest.java | 1 +
.../StatefulParDoEvaluatorFactoryTest.java | 1 +
.../beam/runners/direct/StructuralKeyTest.java | 1 +
.../runners/direct/WatermarkManagerTest.java | 1 +
runners/local-java/pom.xml | 13 +++
.../apache/beam/runners/core/local/Bundle.java | 33 ------
.../beam/runners/core/local/package-info.java | 22 ----
.../org/apache/beam/runners/local/Bundle.java | 33 ++++++
.../beam/runners/local/StructuralKey.java | 103 +++++++++++++++++++
.../apache/beam/runners/local/package-info.java | 22 ++++
.../beam/runners/local/StructuralKeyTest.java | 88 ++++++++++++++++
36 files changed, 288 insertions(+), 159 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/direct-java/pom.xml b/runners/direct-java/pom.xml
index ff7b4a8..65ebe92 100644
--- a/runners/direct-java/pom.xml
+++ b/runners/direct-java/pom.xml
@@ -61,6 +61,7 @@
<include>org.apache.beam:beam-model-pipeline</include>
<include>org.apache.beam:beam-runners-core-construction-java</include>
<include>org.apache.beam:beam-runners-core-java</include>
+ <include>org.apache.beam:beam-runners-core-local-java</include>
<include>com.google.code.findbugs:jsr305</include>
</includes>
</artifactSet>
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BundleFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BundleFactory.java
index e39b5d8..92f9515 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BundleFactory.java
@@ -18,6 +18,7 @@
package org.apache.beam.runners.direct;
import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupByKeyOnly;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CloningBundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CloningBundleFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CloningBundleFactory.java
index 68b059f..502004f 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CloningBundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CloningBundleFactory.java
@@ -18,6 +18,7 @@
package org.apache.beam.runners.direct;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.util.CoderUtils;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
index ccb031f..07c3d22 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CommittedBundle.java
@@ -20,7 +20,8 @@ package org.apache.beam.runners.direct;
import javax.annotation.Nullable;
import org.apache.beam.runners.core.TimerInternals.TimerData;
-import org.apache.beam.runners.core.local.Bundle;
+import org.apache.beam.runners.local.Bundle;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectExecutionContext.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectExecutionContext.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectExecutionContext.java
index e8ad8d7..ef0fc89 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectExecutionContext.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectExecutionContext.java
@@ -23,6 +23,7 @@ import org.apache.beam.runners.core.StepContext;
import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
+import org.apache.beam.runners.local.StructuralKey;
/**
* Execution Context for the {@link DirectRunner}.
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
index d192785..5cdd76d 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
@@ -39,6 +39,7 @@ import org.apache.beam.runners.direct.CommittedResult.OutputType;
import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupByKeyOnly;
import org.apache.beam.runners.direct.WatermarkManager.FiredTimers;
import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.PTransform;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
index d80e4ff..a0d19bf 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
@@ -38,6 +38,7 @@ import org.apache.beam.runners.core.triggers.ExecutableTriggerStateMachine;
import org.apache.beam.runners.core.triggers.TriggerStateMachines;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupAlsoByWindow;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactory.java
index 08623ef..1f77fba 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactory.java
@@ -32,6 +32,7 @@ import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupByKeyOnly;
import org.apache.beam.runners.direct.StepTransformResult.Builder;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.KvCoder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
index 9aabddc..2530e4e 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactory.java
@@ -22,6 +22,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.SetMultimap;
import org.apache.beam.runners.direct.DirectRunner.Enforcement;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.transforms.DoFn;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
index bea7699..1cb0fc3 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
@@ -25,6 +25,7 @@ import com.google.common.collect.ImmutableList;
import java.util.Iterator;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/MultiStepCombine.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/MultiStepCombine.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/MultiStepCombine.java
index 5253ef5..868a3f1 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/MultiStepCombine.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/MultiStepCombine.java
@@ -35,6 +35,7 @@ import org.apache.beam.runners.core.construction.CombineTranslation;
import org.apache.beam.runners.core.construction.PTransformTranslation;
import org.apache.beam.runners.core.construction.PTransformTranslation.RawPTransform;
import org.apache.beam.runners.core.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.CannotProvideCoderException;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.KvCoder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
index 26da6c6..e1d5c6b 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
@@ -31,6 +31,7 @@ import org.apache.beam.runners.core.ReadyCheckingSideInputReader;
import org.apache.beam.runners.core.SimplePushbackSideInputDoFnRunner;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
index 47df0d4..ef06251 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.Map;
import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
index 42bfe0b..257a256 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
@@ -38,6 +38,7 @@ import org.apache.beam.runners.core.StateTags;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
import org.apache.beam.runners.direct.ParDoMultiOverrideFactory.StatefulParDo;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.state.StateSpec;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepAndKey.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepAndKey.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepAndKey.java
index 5440dde..2f2d18e 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepAndKey.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepAndKey.java
@@ -19,6 +19,7 @@ package org.apache.beam.runners.direct;
import com.google.common.base.MoreObjects;
import java.util.Objects;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.runners.AppliedPTransform;
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StructuralKey.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StructuralKey.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StructuralKey.java
deleted file mode 100644
index 0d0f8e7..0000000
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StructuralKey.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * 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.direct;
-
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.coders.CoderException;
-import org.apache.beam.sdk.util.CoderUtils;
-
-/**
- * A (Key, Coder) pair that uses the structural value of the key (as provided by
- * {@link Coder#structuralValue(Object)}) to perform equality and hashing.
- */
-abstract class StructuralKey<K> {
-
- private StructuralKey() {
- // Prevents extending outside of this class
- }
-
- /**
- * Returns the key that this {@link StructuralKey} was created from.
- */
- public abstract K getKey();
-
- /**
- * Get the empty {@link StructuralKey}. All instances of the empty key are considered equal.
- */
- static StructuralKey<?> empty() {
- StructuralKey<Object> emptyKey = new StructuralKey<Object>() {
- @Override
- public Object getKey() {
- return this;
- }
- };
- return emptyKey;
- }
-
- /**
- * Create a new Structural Key of the provided key that can be encoded by the provided coder.
- */
- static <K> StructuralKey<K> of(K key, Coder<K> coder) {
- try {
- return new CoderStructuralKey<>(coder, key);
- } catch (Exception e) {
- throw new IllegalArgumentException(
- "Could not encode a key with its provided coder " + coder.getClass().getSimpleName(), e);
- }
- }
-
- private static class CoderStructuralKey<K> extends StructuralKey<K> {
- private final Coder<K> coder;
- private final Object structuralValue;
- private final byte[] encoded;
-
- private CoderStructuralKey(Coder<K> coder, K key) throws Exception {
- this.coder = coder;
- this.structuralValue = coder.structuralValue(key);
- this.encoded = CoderUtils.encodeToByteArray(coder, key);
- }
-
- @Override
- public K getKey() {
- try {
- return CoderUtils.decodeFromByteArray(coder, encoded);
- } catch (CoderException e) {
- throw new IllegalArgumentException(
- "Could not decode Key with coder of type " + coder.getClass().getSimpleName(), e);
- }
- }
-
- @Override
- public boolean equals(Object other) {
- if (other == this) {
- return true;
- }
- if (other instanceof CoderStructuralKey) {
- CoderStructuralKey<?> that = (CoderStructuralKey<?>) other;
- return structuralValue.equals(that.structuralValue);
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return structuralValue.hashCode();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
index 2371d3b..99e08fe 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
@@ -22,6 +22,7 @@ import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.io.Read.Unbounded;
import org.apache.beam.sdk.transforms.PTransform;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
index a1395a9..3bc5b87 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
@@ -55,6 +55,7 @@ import org.apache.beam.runners.core.StateNamespace;
import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.core.construction.TransformInputs;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.state.TimeDomain;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CloningBundleFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CloningBundleFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CloningBundleFactoryTest.java
index 5bc8077..77d0748 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CloningBundleFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CloningBundleFactoryTest.java
@@ -31,6 +31,7 @@ import com.google.common.collect.Iterables;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.AtomicCoder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.KvCoder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
index f80515f..c96d706 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectTimerInternalsTest.java
@@ -27,6 +27,7 @@ import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate.TimerUpdateBuilder;
import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.state.TimeDomain;
import org.joda.time.Duration;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
index 0a1ffe7..251acf1 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
@@ -39,6 +39,7 @@ import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
import org.apache.beam.runners.direct.WatermarkManager.FiredTimers;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.IterableCoder;
import org.apache.beam.sdk.coders.KvCoder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
index 6dcd5e2..aeee220 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
@@ -28,6 +28,7 @@ import com.google.common.collect.Multiset;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupByKeyOnly;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
index 1373219..0f58150 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
@@ -28,6 +28,7 @@ import com.google.common.collect.Multiset;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupByKeyOnly;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactoryTest.java
index 95c0ad1..f2c9e3c 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityCheckingBundleFactoryTest.java
@@ -20,6 +20,7 @@ package org.apache.beam.runners.direct;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.Assert.assertThat;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.testing.TestPipeline;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
index 83426c6..d2d2317 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
@@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
index 7912538..8c8060f 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
@@ -33,6 +33,7 @@ import javax.annotation.Nullable;
import org.apache.beam.runners.core.ReadyCheckingSideInputReader;
import org.apache.beam.runners.direct.DirectExecutionContext.DirectStepContext;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
index fe0b743..fb45bd5 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
@@ -44,6 +44,7 @@ import org.apache.beam.runners.core.StateTags;
import org.apache.beam.runners.core.construction.TransformInputs;
import org.apache.beam.runners.direct.ParDoMultiOverrideFactory.StatefulParDo;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.runners.AppliedPTransform;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StructuralKeyTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StructuralKeyTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StructuralKeyTest.java
index 9840c72..59ff88f 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StructuralKeyTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StructuralKeyTest.java
@@ -24,6 +24,7 @@ import static org.hamcrest.Matchers.theInstance;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertThat;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
index e3f6215..79035e6 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
@@ -41,6 +41,7 @@ import org.apache.beam.runners.direct.WatermarkManager.FiredTimers;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate.TimerUpdateBuilder;
import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
+import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/local-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/local-java/pom.xml b/runners/local-java/pom.xml
index b988fd9..e1ac6f8 100644
--- a/runners/local-java/pom.xml
+++ b/runners/local-java/pom.xml
@@ -58,5 +58,18 @@
<artifactId>auto-value</artifactId>
<scope>provided</scope>
</dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-all</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java b/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java
deleted file mode 100644
index 98e1e8a..0000000
--- a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/Bundle.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.local;
-
-import org.apache.beam.sdk.util.WindowedValue;
-import org.joda.time.Instant;
-
-/** An immutable collection of elements which are part of a {@code PCollection}. */
-public interface Bundle<T> extends Iterable<WindowedValue<T>> {
- /**
- * Return the minimum timestamp among elements in this bundle.
- *
- * <p>This should be equivalent to iterating over all of the elements within a bundle and
- * selecting the minimum timestamp from among them.
- */
- Instant getMinimumTimestamp();
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java b/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java
deleted file mode 100644
index ffd2e54..0000000
--- a/runners/local-java/src/main/java/org/apache/beam/runners/core/local/package-info.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.
- */
-
-/**
- * Utilities useful when executing a pipeline on a single machine.
- */
-package org.apache.beam.runners.core.local;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/local-java/src/main/java/org/apache/beam/runners/local/Bundle.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/main/java/org/apache/beam/runners/local/Bundle.java b/runners/local-java/src/main/java/org/apache/beam/runners/local/Bundle.java
new file mode 100644
index 0000000..959494b
--- /dev/null
+++ b/runners/local-java/src/main/java/org/apache/beam/runners/local/Bundle.java
@@ -0,0 +1,33 @@
+/*
+ * 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.local;
+
+import org.apache.beam.sdk.util.WindowedValue;
+import org.joda.time.Instant;
+
+/** An immutable collection of elements which are part of a {@code PCollection}. */
+public interface Bundle<T> extends Iterable<WindowedValue<T>> {
+ /**
+ * Return the minimum timestamp among elements in this bundle.
+ *
+ * <p>This should be equivalent to iterating over all of the elements within a bundle and
+ * selecting the minimum timestamp from among them.
+ */
+ Instant getMinimumTimestamp();
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/local-java/src/main/java/org/apache/beam/runners/local/StructuralKey.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/main/java/org/apache/beam/runners/local/StructuralKey.java b/runners/local-java/src/main/java/org/apache/beam/runners/local/StructuralKey.java
new file mode 100644
index 0000000..ac863da
--- /dev/null
+++ b/runners/local-java/src/main/java/org/apache/beam/runners/local/StructuralKey.java
@@ -0,0 +1,103 @@
+/*
+ * 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.local;
+
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.util.CoderUtils;
+
+/**
+ * A (Key, Coder) pair that uses the structural value of the key (as provided by
+ * {@link Coder#structuralValue(Object)}) to perform equality and hashing.
+ */
+public abstract class StructuralKey<K> {
+
+ private StructuralKey() {
+ // Prevents extending outside of this class
+ }
+
+ /**
+ * Returns the key that this {@link StructuralKey} was created from.
+ */
+ public abstract K getKey();
+
+ /**
+ * Get the empty {@link StructuralKey}. All instances of the empty key are considered equal.
+ */
+ public static StructuralKey<?> empty() {
+ StructuralKey<Object> emptyKey = new StructuralKey<Object>() {
+ @Override
+ public Object getKey() {
+ return this;
+ }
+ };
+ return emptyKey;
+ }
+
+ /**
+ * Create a new Structural Key of the provided key that can be encoded by the provided coder.
+ */
+ public static <K> StructuralKey<K> of(K key, Coder<K> coder) {
+ try {
+ return new CoderStructuralKey<>(coder, key);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ "Could not encode a key with its provided coder " + coder.getClass().getSimpleName(), e);
+ }
+ }
+
+ private static class CoderStructuralKey<K> extends StructuralKey<K> {
+ private final Coder<K> coder;
+ private final Object structuralValue;
+ private final byte[] encoded;
+
+ private CoderStructuralKey(Coder<K> coder, K key) throws Exception {
+ this.coder = coder;
+ this.structuralValue = coder.structuralValue(key);
+ this.encoded = CoderUtils.encodeToByteArray(coder, key);
+ }
+
+ @Override
+ public K getKey() {
+ try {
+ return CoderUtils.decodeFromByteArray(coder, encoded);
+ } catch (CoderException e) {
+ throw new IllegalArgumentException(
+ "Could not decode Key with coder of type " + coder.getClass().getSimpleName(), e);
+ }
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == this) {
+ return true;
+ }
+ if (other instanceof CoderStructuralKey) {
+ CoderStructuralKey<?> that = (CoderStructuralKey<?>) other;
+ return structuralValue.equals(that.structuralValue);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return structuralValue.hashCode();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/local-java/src/main/java/org/apache/beam/runners/local/package-info.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/main/java/org/apache/beam/runners/local/package-info.java b/runners/local-java/src/main/java/org/apache/beam/runners/local/package-info.java
new file mode 100644
index 0000000..c7d438e
--- /dev/null
+++ b/runners/local-java/src/main/java/org/apache/beam/runners/local/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Utilities useful when executing a pipeline on a single machine.
+ */
+package org.apache.beam.runners.local;
http://git-wip-us.apache.org/repos/asf/beam/blob/c8d45d4a/runners/local-java/src/test/java/org/apache/beam/runners/local/StructuralKeyTest.java
----------------------------------------------------------------------
diff --git a/runners/local-java/src/test/java/org/apache/beam/runners/local/StructuralKeyTest.java b/runners/local-java/src/test/java/org/apache/beam/runners/local/StructuralKeyTest.java
new file mode 100644
index 0000000..b0b6112
--- /dev/null
+++ b/runners/local-java/src/test/java/org/apache/beam/runners/local/StructuralKeyTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.local;
+
+import static org.junit.Assert.assertArrayEquals;
+
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.hamcrest.Matchers;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link StructuralKey}.
+ */
+@RunWith(JUnit4.class)
+public class StructuralKeyTest {
+ @Test
+ public void getKeyEqualToOldKey() {
+ Assert.assertThat(StructuralKey.of(1234, VarIntCoder.of()).getKey(), Matchers.equalTo(1234));
+ Assert.assertThat(
+ StructuralKey.of("foobar", StringUtf8Coder.of()).getKey(), Matchers.equalTo("foobar"));
+ assertArrayEquals(StructuralKey.of(new byte[] {2, 9, -22}, ByteArrayCoder.of()).getKey(),
+ new byte[] {2, 9, -22});
+ }
+
+ @Test
+ public void getKeyNotSameInstance() {
+ byte[] original = new byte[] {1, 4, 9, 127, -22};
+ StructuralKey<byte[]> key = StructuralKey.of(original, ByteArrayCoder.of());
+
+ Assert.assertThat(key.getKey(), Matchers.not(Matchers.theInstance(original)));
+ }
+
+ @Test
+ public void emptyKeysNotEqual() {
+ StructuralKey<?> empty = StructuralKey.empty();
+
+ Assert.assertThat(
+ empty, Matchers.not(Matchers.<StructuralKey<?>>equalTo(StructuralKey.empty())));
+ Assert.assertThat(empty, Matchers.<StructuralKey<?>>equalTo(empty));
+ }
+
+ @Test
+ public void objectEqualsTrueKeyEquals() {
+ StructuralKey<Integer> original = StructuralKey.of(1234, VarIntCoder.of());
+ Assert.assertThat(StructuralKey.of(1234, VarIntCoder.of()), Matchers.equalTo(original));
+ }
+
+ @Test
+ public void objectsNotEqualEncodingsEqualEquals() {
+ byte[] original = new byte[] {1, 4, 9, 127, -22};
+ StructuralKey<byte[]> key = StructuralKey.of(original, ByteArrayCoder.of());
+
+ StructuralKey<byte[]> otherKey =
+ StructuralKey.of(new byte[] {1, 4, 9, 127, -22}, ByteArrayCoder.of());
+ Assert.assertThat(key, Matchers.equalTo(otherKey));
+ }
+
+ @Test
+ public void notEqualEncodingsEqual() {
+ byte[] original = new byte[] {1, 4, 9, 127, -22};
+ StructuralKey<byte[]> key = StructuralKey.of(original, ByteArrayCoder.of());
+
+ StructuralKey<byte[]> otherKey =
+ StructuralKey.of(new byte[] {9, -128, 22}, ByteArrayCoder.of());
+ Assert.assertThat(key, Matchers.not(Matchers.equalTo(otherKey)));
+ }
+}