beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bchamb...@apache.org
Subject [1/2] incubator-beam git commit: Remove InProcessBundle
Date Tue, 19 Apr 2016 16:02:58 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 861562239 -> 442435ed0


Remove InProcessBundle

InProcessBundle is an implementation detail of InProcessBundleFactory,
which it should be contained within.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/3987a05b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/3987a05b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/3987a05b

Branch: refs/heads/master
Commit: 3987a05b83de46823efde2742d585cb5b7b98cc3
Parents: 8615622
Author: Thomas Groh <tgroh@google.com>
Authored: Mon Apr 18 12:55:26 2016 -0700
Committer: bchambers <bchambers@google.com>
Committed: Tue Apr 19 08:49:39 2016 -0700

----------------------------------------------------------------------
 .../sdk/runners/inprocess/InProcessBundle.java  | 124 ----------------
 .../EncodabilityEnforcementFactoryTest.java     |  11 +-
 .../ImmutabilityEnforcementFactoryTest.java     |   8 +-
 .../inprocess/InMemoryWatermarkManagerTest.java |   2 +-
 .../runners/inprocess/InProcessBundleTest.java  | 145 -------------------
 .../InProcessEvaluationContextTest.java         |  26 +---
 .../inprocess/TransformExecutorTest.java        |   6 +-
 7 files changed, 23 insertions(+), 299 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3987a05b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/InProcessBundle.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/InProcessBundle.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/InProcessBundle.java
deleted file mode 100644
index d3da9e1..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/InProcessBundle.java
+++ /dev/null
@@ -1,124 +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.sdk.runners.inprocess;
-
-import static com.google.common.base.Preconditions.checkState;
-
-import org.apache.beam.sdk.runners.inprocess.InProcessPipelineRunner.CommittedBundle;
-import org.apache.beam.sdk.runners.inprocess.InProcessPipelineRunner.UncommittedBundle;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.values.PCollection;
-
-import com.google.common.base.MoreObjects;
-import com.google.common.collect.ImmutableList;
-
-import org.joda.time.Instant;
-
-import javax.annotation.Nullable;
-
-/**
- * A {@link UncommittedBundle} that buffers elements in memory.
- */
-public final class InProcessBundle<T> implements UncommittedBundle<T> {
-  private final PCollection<T> pcollection;
-  private final boolean keyed;
-  private final Object key;
-  private boolean committed = false;
-  private ImmutableList.Builder<WindowedValue<T>> elements;
-
-  /**
-   * Create a new {@link InProcessBundle} for the specified {@link PCollection} without a
key.
-   */
-  public static <T> InProcessBundle<T> unkeyed(PCollection<T> pcollection)
{
-    return new InProcessBundle<T>(pcollection, false, null);
-  }
-
-  /**
-   * Create a new {@link InProcessBundle} for the specified {@link PCollection} with the
specified
-   * key.
-   *
-   * See {@link CommittedBundle#getKey()} and {@link CommittedBundle#isKeyed()} for more
-   * information.
-   */
-  public static <T> InProcessBundle<T> keyed(PCollection<T> pcollection,
Object key) {
-    return new InProcessBundle<T>(pcollection, true, key);
-  }
-
-  private InProcessBundle(PCollection<T> pcollection, boolean keyed, Object key) {
-    this.pcollection = pcollection;
-    this.keyed = keyed;
-    this.key = key;
-    this.elements = ImmutableList.builder();
-  }
-
-  @Override
-  public PCollection<T> getPCollection() {
-    return pcollection;
-  }
-
-  @Override
-  public InProcessBundle<T> add(WindowedValue<T> element) {
-    checkState(!committed, "Can't add element %s to committed bundle %s", element, this);
-    elements.add(element);
-    return this;
-  }
-
-  @Override
-  public CommittedBundle<T> commit(final Instant synchronizedCompletionTime) {
-    checkState(!committed, "Can't commit already committed bundle %s", this);
-    committed = true;
-    final Iterable<WindowedValue<T>> committedElements = elements.build();
-    return new CommittedBundle<T>() {
-      @Override
-      @Nullable
-      public Object getKey() {
-        return key;
-      }
-
-      @Override
-      public boolean isKeyed() {
-        return keyed;
-      }
-
-      @Override
-      public Iterable<WindowedValue<T>> getElements() {
-        return committedElements;
-      }
-
-      @Override
-      public PCollection<T> getPCollection() {
-        return pcollection;
-      }
-
-      @Override
-      public Instant getSynchronizedProcessingOutputWatermark() {
-        return synchronizedCompletionTime;
-      }
-
-      @Override
-      public String toString() {
-        return MoreObjects.toStringHelper(this)
-            .omitNullValues()
-            .add("pcollection", pcollection)
-            .add("key", key)
-            .add("elements", committedElements)
-            .toString();
-      }
-    };
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3987a05b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
index 7720589..85c4322 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
@@ -50,6 +50,7 @@ import java.util.Collections;
 public class EncodabilityEnforcementFactoryTest {
   @Rule public ExpectedException thrown = ExpectedException.none();
   private EncodabilityEnforcementFactory factory = EncodabilityEnforcementFactory.create();
+  private BundleFactory bundleFactory = InProcessBundleFactory.create();
 
   @Test
   public void encodeFailsThrows() {
@@ -61,7 +62,7 @@ public class EncodabilityEnforcementFactoryTest {
 
     WindowedValue<Record> record = WindowedValue.valueInGlobalWindow(new Record());
     CommittedBundle<Record> input =
-        InProcessBundle.unkeyed(unencodable).add(record).commit(Instant.now());
+        bundleFactory.createRootBundle(unencodable).add(record).commit(Instant.now());
     ModelEnforcement<Record> enforcement = factory.forBundle(input, consumer);
 
     thrown.expect(UserCodeException.class);
@@ -80,7 +81,7 @@ public class EncodabilityEnforcementFactoryTest {
     WindowedValue<Record> record = WindowedValue.valueInGlobalWindow(new Record());
 
     CommittedBundle<Record> input =
-        InProcessBundle.unkeyed(unencodable).add(record).commit(Instant.now());
+        bundleFactory.createRootBundle(unencodable).add(record).commit(Instant.now());
     ModelEnforcement<Record> enforcement = factory.forBundle(input, consumer);
 
     thrown.expect(UserCodeException.class);
@@ -107,7 +108,7 @@ public class EncodabilityEnforcementFactoryTest {
             });
 
     CommittedBundle<Record> input =
-        InProcessBundle.unkeyed(unencodable).add(record).commit(Instant.now());
+        bundleFactory.createRootBundle(unencodable).add(record).commit(Instant.now());
     ModelEnforcement<Record> enforcement = factory.forBundle(input, consumer);
 
     thrown.expect(UserCodeException.class);
@@ -131,7 +132,7 @@ public class EncodabilityEnforcementFactoryTest {
     WindowedValue<Record> record = WindowedValue.<Record>valueInGlobalWindow(new
Record());
 
     CommittedBundle<Record> input =
-        InProcessBundle.unkeyed(unencodable).add(record).commit(Instant.now());
+        bundleFactory.createRootBundle(unencodable).add(record).commit(Instant.now());
     ModelEnforcement<Record> enforcement = factory.forBundle(input, consumer);
 
     enforcement.beforeElement(record);
@@ -152,7 +153,7 @@ public class EncodabilityEnforcementFactoryTest {
     WindowedValue<Integer> value = WindowedValue.valueInGlobalWindow(1);
 
     CommittedBundle<Integer> input =
-        InProcessBundle.unkeyed(unencodable).add(value).commit(Instant.now());
+        bundleFactory.createRootBundle(unencodable).add(value).commit(Instant.now());
     ModelEnforcement<Integer> enforcement = factory.forBundle(input, consumer);
 
     enforcement.beforeElement(value);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3987a05b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityEnforcementFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityEnforcementFactoryTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityEnforcementFactoryTest.java
index 4520504..16633ed 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityEnforcementFactoryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityEnforcementFactoryTest.java
@@ -46,12 +46,14 @@ import java.util.Collections;
 public class ImmutabilityEnforcementFactoryTest implements Serializable {
   @Rule public transient ExpectedException thrown = ExpectedException.none();
   private transient ImmutabilityEnforcementFactory factory;
+  private transient BundleFactory bundleFactory;
   private transient PCollection<byte[]> pcollection;
   private transient AppliedPTransform<?, ?, ?> consumer;
 
   @Before
   public void setup() {
     factory = new ImmutabilityEnforcementFactory();
+    bundleFactory = InProcessBundleFactory.create();
     TestPipeline p = TestPipeline.create();
     pcollection =
         p.apply(Create.of("foo".getBytes(), "spamhameggs".getBytes()))
@@ -71,7 +73,7 @@ public class ImmutabilityEnforcementFactoryTest implements Serializable
{
   public void unchangedSucceeds() {
     WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes());
     CommittedBundle<byte[]> elements =
-        InProcessBundle.unkeyed(pcollection).add(element).commit(Instant.now());
+        bundleFactory.createRootBundle(pcollection).add(element).commit(Instant.now());
 
     ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer);
     enforcement.beforeElement(element);
@@ -86,7 +88,7 @@ public class ImmutabilityEnforcementFactoryTest implements Serializable
{
   public void mutatedDuringProcessElementThrows() {
     WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes());
     CommittedBundle<byte[]> elements =
-        InProcessBundle.unkeyed(pcollection).add(element).commit(Instant.now());
+        bundleFactory.createRootBundle(pcollection).add(element).commit(Instant.now());
 
     ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer);
     enforcement.beforeElement(element);
@@ -107,7 +109,7 @@ public class ImmutabilityEnforcementFactoryTest implements Serializable
{
 
     WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes());
     CommittedBundle<byte[]> elements =
-        InProcessBundle.unkeyed(pcollection).add(element).commit(Instant.now());
+        bundleFactory.createRootBundle(pcollection).add(element).commit(Instant.now());
 
     ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer);
     enforcement.beforeElement(element);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3987a05b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InMemoryWatermarkManagerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InMemoryWatermarkManagerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InMemoryWatermarkManagerTest.java
index 077c0e7..736076c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InMemoryWatermarkManagerTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InMemoryWatermarkManagerTest.java
@@ -798,7 +798,7 @@ public class InMemoryWatermarkManagerTest implements Serializable {
 
     Instant upstreamHold = new Instant(2048L);
     CommittedBundle<Integer> filteredBundle =
-        bundleFactory.createKeyedBundle(null, "key", filtered).commit(upstreamHold);
+        bundleFactory.createKeyedBundle(created, "key", filtered).commit(upstreamHold);
     manager.updateWatermarks(
         created,
         filtered.getProducingTransformInternal(),

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3987a05b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessBundleTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessBundleTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessBundleTest.java
deleted file mode 100644
index 103ace5..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessBundleTest.java
+++ /dev/null
@@ -1,145 +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.sdk.runners.inprocess;
-
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.Assert.assertThat;
-
-import org.apache.beam.sdk.runners.inprocess.InProcessPipelineRunner.CommittedBundle;
-import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.values.PCollection;
-
-import com.google.common.collect.ImmutableList;
-
-import org.hamcrest.Matcher;
-import org.hamcrest.Matchers;
-import org.joda.time.Instant;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-
-/**
- * Tests for {@link InProcessBundle}.
- */
-@RunWith(JUnit4.class)
-public class InProcessBundleTest {
-  @Rule
-  public ExpectedException thrown = ExpectedException.none();
-
-  @Test
-  public void unkeyedShouldCreateWithNullKey() {
-    PCollection<Integer> pcollection = TestPipeline.create().apply(Create.of(1));
-
-    InProcessBundle<Integer> inFlightBundle = InProcessBundle.unkeyed(pcollection);
-
-    CommittedBundle<Integer> bundle = inFlightBundle.commit(Instant.now());
-
-    assertThat(bundle.isKeyed(), is(false));
-    assertThat(bundle.getKey(), nullValue());
-  }
-
-  private void keyedCreateBundle(Object key) {
-    PCollection<Integer> pcollection = TestPipeline.create().apply(Create.of(1));
-
-    InProcessBundle<Integer> inFlightBundle = InProcessBundle.keyed(pcollection, key);
-
-    CommittedBundle<Integer> bundle = inFlightBundle.commit(Instant.now());
-    assertThat(bundle.isKeyed(), is(true));
-    assertThat(bundle.getKey(), equalTo(key));
-  }
-
-  @Test
-  public void keyedWithNullKeyShouldCreateKeyedBundle() {
-    keyedCreateBundle(null);
-  }
-
-  @Test
-  public void keyedWithKeyShouldCreateKeyedBundle() {
-    keyedCreateBundle(new Object());
-  }
-
-  private <T> void afterCommitGetElementsShouldHaveAddedElements(Iterable<WindowedValue<T>>
elems) {
-    PCollection<T> pcollection = TestPipeline.create().apply(Create.<T>of());
-
-    InProcessBundle<T> bundle = InProcessBundle.unkeyed(pcollection);
-    Collection<Matcher<? super WindowedValue<T>>> expectations = new ArrayList<>();
-    for (WindowedValue<T> elem : elems) {
-      bundle.add(elem);
-      expectations.add(equalTo(elem));
-    }
-    Matcher<Iterable<? extends WindowedValue<T>>> containsMatcher =
-        Matchers.<WindowedValue<T>>containsInAnyOrder(expectations);
-    assertThat(bundle.commit(Instant.now()).getElements(), containsMatcher);
-  }
-
-  @Test
-  public void getElementsBeforeAddShouldReturnEmptyIterable() {
-    afterCommitGetElementsShouldHaveAddedElements(Collections.<WindowedValue<Integer>>emptyList());
-  }
-
-  @Test
-  public void getElementsAfterAddShouldReturnAddedElements() {
-    WindowedValue<Integer> firstValue = WindowedValue.valueInGlobalWindow(1);
-    WindowedValue<Integer> secondValue =
-        WindowedValue.timestampedValueInGlobalWindow(2, new Instant(1000L));
-
-    afterCommitGetElementsShouldHaveAddedElements(ImmutableList.of(firstValue, secondValue));
-  }
-
-  @Test
-  public void addAfterCommitShouldThrowException() {
-    PCollection<Integer> pcollection = TestPipeline.create().apply(Create.<Integer>of());
-
-    InProcessBundle<Integer> bundle = InProcessBundle.unkeyed(pcollection);
-    bundle.add(WindowedValue.valueInGlobalWindow(1));
-    CommittedBundle<Integer> firstCommit = bundle.commit(Instant.now());
-    assertThat(firstCommit.getElements(), containsInAnyOrder(WindowedValue.valueInGlobalWindow(1)));
-
-    thrown.expect(IllegalStateException.class);
-    thrown.expectMessage("3");
-    thrown.expectMessage("committed");
-
-    bundle.add(WindowedValue.valueInGlobalWindow(3));
-  }
-
-  @Test
-  public void commitAfterCommitShouldThrowException() {
-    PCollection<Integer> pcollection = TestPipeline.create().apply(Create.<Integer>of());
-
-    InProcessBundle<Integer> bundle = InProcessBundle.unkeyed(pcollection);
-    bundle.add(WindowedValue.valueInGlobalWindow(1));
-    CommittedBundle<Integer> firstCommit = bundle.commit(Instant.now());
-    assertThat(firstCommit.getElements(), containsInAnyOrder(WindowedValue.valueInGlobalWindow(1)));
-
-    thrown.expect(IllegalStateException.class);
-    thrown.expectMessage("committed");
-
-    bundle.commit(Instant.now());
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3987a05b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessEvaluationContextTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessEvaluationContextTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessEvaluationContextTest.java
index 6736562..50b83fd 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessEvaluationContextTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/InProcessEvaluationContextTest.java
@@ -93,6 +93,8 @@ public class InProcessEvaluationContextTest {
   private Collection<AppliedPTransform<?, ?, ?>> rootTransforms;
   private Map<PValue, Collection<AppliedPTransform<?, ?, ?>>> valueToConsumers;
 
+  private BundleFactory bundleFactory;
+
   @Before
   public void setup() {
     InProcessPipelineRunner runner =
@@ -110,6 +112,8 @@ public class InProcessEvaluationContextTest {
     rootTransforms = cVis.getRootTransforms();
     valueToConsumers = cVis.getValueToConsumers();
 
+    bundleFactory = InProcessBundleFactory.create();
+
     context =
         InProcessEvaluationContext.create(
             runner.getPipelineOptions(),
@@ -393,37 +397,23 @@ public class InProcessEvaluationContextTest {
   }
 
   @Test
-  public void createBundleUnkeyedResultUnkeyed() {
-    CommittedBundle<KV<String, Integer>> newBundle =
-        context
-            .createBundle(InProcessBundle.unkeyed(created).commit(Instant.now()), downstream)
-            .commit(Instant.now());
-    assertThat(newBundle.isKeyed(), is(false));
-  }
-
-  @Test
   public void createBundleKeyedResultPropagatesKey() {
     CommittedBundle<KV<String, Integer>> newBundle =
         context
-            .createBundle(InProcessBundle.keyed(created, "foo").commit(Instant.now()), downstream)
+            .createBundle(
+                bundleFactory.createKeyedBundle(null, "foo", created).commit(Instant.now()),
+                downstream)
             .commit(Instant.now());
-    assertThat(newBundle.isKeyed(), is(true));
     assertThat(newBundle.getKey(), Matchers.<Object>equalTo("foo"));
   }
 
   @Test
-  public void createRootBundleUnkeyed() {
-    assertThat(context.createRootBundle(created).commit(Instant.now()).isKeyed(), is(false));
-  }
-
-  @Test
   public void createKeyedBundleKeyed() {
     CommittedBundle<KV<String, Integer>> keyedBundle =
         context
             .createKeyedBundle(
-                InProcessBundle.unkeyed(created).commit(Instant.now()), "foo", downstream)
+                bundleFactory.createRootBundle(created).commit(Instant.now()), "foo", downstream)
             .commit(Instant.now());
-    assertThat(keyedBundle.isKeyed(), is(true));
     assertThat(keyedBundle.getKey(), Matchers.<Object>equalTo("foo"));
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3987a05b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/TransformExecutorTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/TransformExecutorTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/TransformExecutorTest.java
index d29ffac..d3d70e0 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/TransformExecutorTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/TransformExecutorTest.java
@@ -332,7 +332,7 @@ public class TransformExecutorTest {
     WindowedValue<String> fooElem = WindowedValue.valueInGlobalWindow("foo");
     WindowedValue<String> barElem = WindowedValue.valueInGlobalWindow("bar");
     CommittedBundle<String> inputBundle =
-        InProcessBundle.unkeyed(created).add(fooElem).add(barElem).commit(Instant.now());
+        bundleFactory.createRootBundle(created).add(fooElem).add(barElem).commit(Instant.now());
     when(
             registry.forApplication(
                 downstream.getProducingTransformInternal(), inputBundle, evaluationContext))
@@ -393,7 +393,7 @@ public class TransformExecutorTest {
 
     WindowedValue<byte[]> fooBytes = WindowedValue.valueInGlobalWindow("foo".getBytes());
     CommittedBundle<byte[]> inputBundle =
-        InProcessBundle.unkeyed(pcBytes).add(fooBytes).commit(Instant.now());
+        bundleFactory.createRootBundle(pcBytes).add(fooBytes).commit(Instant.now());
     when(
             registry.forApplication(
                 pcBytes.getProducingTransformInternal(), inputBundle, evaluationContext))
@@ -452,7 +452,7 @@ public class TransformExecutorTest {
 
     WindowedValue<byte[]> fooBytes = WindowedValue.valueInGlobalWindow("foo".getBytes());
     CommittedBundle<byte[]> inputBundle =
-        InProcessBundle.unkeyed(pcBytes).add(fooBytes).commit(Instant.now());
+        bundleFactory.createRootBundle(pcBytes).add(fooBytes).commit(Instant.now());
     when(
             registry.forApplication(
                 pcBytes.getProducingTransformInternal(), inputBundle, evaluationContext))


Mime
View raw message