You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2018/06/07 20:17:47 UTC
[beam] branch master updated (a6536f9 -> 774ff6a)
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git.
from a6536f9 Merge pull request #5476: Fix FileSystems.match() for local paths on Windows.
new 61a8b10 BEAM-3876 avoid NPE if checkpoint is null in an unbounded source
new 8b7f5eb testing npe fix and exception rethrow in unit tests
new cba4cac Update https://github.com/apache/beam/pull/4894 to correspond with changes to HEAD branch.
new 774ff6a [BEAM-3876] avoid NPE if checkpoint is null in an unbounded source
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../direct/UnboundedReadEvaluatorFactory.java | 29 ++++++++--
.../direct/UnboundedReadEvaluatorFactoryTest.java | 61 ++++++++++++++++++++++
2 files changed, 86 insertions(+), 4 deletions(-)
--
To stop receiving notification emails like this one, please contact
lcwik@apache.org.
[beam] 03/04: Update https://github.com/apache/beam/pull/4894 to
correspond with changes to HEAD branch.
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit cba4cac8e57aaf3bed5bb7b37c744af3d9b3d5d1
Author: Luke Cwik <lc...@google.com>
AuthorDate: Thu Jun 7 13:15:51 2018 -0700
Update https://github.com/apache/beam/pull/4894 to correspond with changes to HEAD branch.
---
.../direct/UnboundedReadEvaluatorFactoryTest.java | 17 ++++++++++-------
1 file changed, 10 insertions(+), 7 deletions(-)
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
index f708024..fe47e77 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
@@ -52,6 +52,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Set;
+import java.util.concurrent.Executors;
import javax.annotation.Nullable;
import org.apache.beam.runners.direct.UnboundedReadDeduplicator.NeverDeduplicator;
import org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory.UnboundedSourceShard;
@@ -114,6 +115,7 @@ public class UnboundedReadEvaluatorFactoryTest {
public void setup() {
source = CountingSource.unboundedWithTimestampFn(new LongToInstantFn());
longs = p.apply(Read.from(source));
+ options = PipelineOptionsFactory.create();
context = mock(EvaluationContext.class);
factory = new UnboundedReadEvaluatorFactory(context, options);
@@ -432,14 +434,15 @@ public class UnboundedReadEvaluatorFactoryTest {
}
private void processElement(final TestUnboundedSource<String> source) throws Exception {
- final DirectOptions options = PipelineOptionsFactory.fromArgs()
- .create().as(DirectOptions.class);
final EvaluationContext context = EvaluationContext.create(
- options, MockClock.fromInstant(Instant.now()), CloningBundleFactory.create(),
- DirectGraph.create(emptyMap(), emptyMap(),
- LinkedListMultimap.create(), LinkedListMultimap.create(), emptySet(), emptyMap()),
- emptySet());
- final UnboundedReadEvaluatorFactory factory = new UnboundedReadEvaluatorFactory(context);
+ MockClock.fromInstant(Instant.now()),
+ CloningBundleFactory.create(),
+ DirectGraph.create(
+ emptyMap(), emptyMap(), LinkedListMultimap.create(), emptySet(), emptyMap()),
+ emptySet(),
+ Executors.newCachedThreadPool());
+ final UnboundedReadEvaluatorFactory factory =
+ new UnboundedReadEvaluatorFactory(context, options);
final Read.Unbounded<String> unbounded = Read.from(source);
final Pipeline pipeline = Pipeline.create(options);
--
To stop receiving notification emails like this one, please contact
lcwik@apache.org.
[beam] 01/04: BEAM-3876 avoid NPE if checkpoint is null in an
unbounded source
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 61a8b10dbddf93cb63faa0957d55695ab69ad81b
Author: Romain Manni-Bucau <rm...@gmail.com>
AuthorDate: Mon Mar 19 16:13:00 2018 +0100
BEAM-3876 avoid NPE if checkpoint is null in an unbounded source
---
.../direct/UnboundedReadEvaluatorFactory.java | 26 ++++++++++++++++++----
1 file changed, 22 insertions(+), 4 deletions(-)
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
index 5c3d2f2..a46d657 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
@@ -177,10 +177,28 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory {
watermark)));
} else {
// End of input. Close the reader after finalizing old checkpoint.
- shard.getCheckpoint().finalizeCheckpoint();
- UnboundedReader<?> toClose = reader;
- reader = null; // Avoid double close below in case of an exception.
- toClose.close();
+ // note: can be null for empty datasets so ensure to null check the checkpoint
+ final CheckpointMarkT checkpoint = shard.getCheckpoint();
+ IOException ioe = null;
+ try {
+ if (checkpoint != null) {
+ checkpoint.finalizeCheckpoint();
+ }
+ } catch (final IOException finalizeCheckpointException) {
+ ioe = finalizeCheckpointException;
+ } finally {
+ try {
+ UnboundedReader<?> toClose = reader;
+ reader = null; // Avoid double close below in case of an exception.
+ toClose.close();
+ } catch (final IOException closeEx) {
+ if (ioe != null) {
+ ioe.addSuppressed(closeEx);
+ } else {
+ throw closeEx;
+ }
+ }
+ }
}
}
} catch (IOException e) {
--
To stop receiving notification emails like this one, please contact
lcwik@apache.org.
[beam] 04/04: [BEAM-3876] avoid NPE if checkpoint is null in an
unbounded source
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 774ff6a5e7914df227bbcf941cd473dc5ca97972
Merge: a6536f9 cba4cac
Author: Luke Cwik <lc...@google.com>
AuthorDate: Thu Jun 7 13:17:24 2018 -0700
[BEAM-3876] avoid NPE if checkpoint is null in an unbounded source
This closes #4894
.../direct/UnboundedReadEvaluatorFactory.java | 29 ++++++++--
.../direct/UnboundedReadEvaluatorFactoryTest.java | 61 ++++++++++++++++++++++
2 files changed, 86 insertions(+), 4 deletions(-)
--
To stop receiving notification emails like this one, please contact
lcwik@apache.org.
[beam] 02/04: testing npe fix and exception rethrow in unit tests
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 8b7f5eb6b00df10ec240faee162a5247478b5ee2
Author: Romain Manni-Bucau <rm...@gmail.com>
AuthorDate: Mon Mar 19 18:25:32 2018 +0100
testing npe fix and exception rethrow in unit tests
---
.../direct/UnboundedReadEvaluatorFactory.java | 3 ++
.../direct/UnboundedReadEvaluatorFactoryTest.java | 58 ++++++++++++++++++++++
2 files changed, 61 insertions(+)
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
index a46d657..4640efd 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
@@ -199,6 +199,9 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory {
}
}
}
+ if (ioe != null) {
+ throw ioe;
+ }
}
}
} catch (IOException e) {
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
index af5313a..f708024 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
@@ -18,6 +18,9 @@
package org.apache.beam.runners.direct;
import static com.google.common.base.Preconditions.checkState;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static java.util.Collections.singletonMap;
import static org.apache.beam.runners.direct.DirectGraphs.getProducer;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
@@ -25,6 +28,7 @@ import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
@@ -36,6 +40,7 @@ import com.google.common.collect.ContiguousSet;
import com.google.common.collect.DiscreteDomain;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
+import com.google.common.collect.LinkedListMultimap;
import com.google.common.collect.Range;
import java.io.IOException;
import java.io.InputStream;
@@ -43,16 +48,19 @@ import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Set;
import javax.annotation.Nullable;
import org.apache.beam.runners.direct.UnboundedReadDeduplicator.NeverDeduplicator;
import org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory.UnboundedSourceShard;
+import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.AtomicCoder;
import org.apache.beam.sdk.coders.BigEndianLongCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.io.CountingSource;
import org.apache.beam.sdk.io.Read;
@@ -66,10 +74,13 @@ import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.util.CoderUtils;
import org.apache.beam.sdk.util.VarInt;
import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TupleTag;
import org.hamcrest.Matchers;
import org.joda.time.DateTime;
import org.joda.time.Instant;
@@ -402,6 +413,53 @@ public class UnboundedReadEvaluatorFactoryTest {
evaluator.processElement(shard);
}
+ @Test // before this was throwing a NPE
+ public void emptySource() throws Exception {
+ TestUnboundedSource.readerClosedCount = 0;
+ final TestUnboundedSource<String> source = new TestUnboundedSource<>(StringUtf8Coder.of());
+ source.advanceWatermarkToInfinity = true;
+ processElement(source);
+ assertEquals(1, TestUnboundedSource.readerClosedCount);
+ TestUnboundedSource.readerClosedCount = 0; // reset
+ }
+
+ @Test(expected = IOException.class)
+ public void sourceThrowingException() throws Exception {
+ final TestUnboundedSource<String> source = new TestUnboundedSource<>(StringUtf8Coder.of());
+ source.advanceWatermarkToInfinity = true;
+ source.throwOnClose = true;
+ processElement(source);
+ }
+
+ private void processElement(final TestUnboundedSource<String> source) throws Exception {
+ final DirectOptions options = PipelineOptionsFactory.fromArgs()
+ .create().as(DirectOptions.class);
+ final EvaluationContext context = EvaluationContext.create(
+ options, MockClock.fromInstant(Instant.now()), CloningBundleFactory.create(),
+ DirectGraph.create(emptyMap(), emptyMap(),
+ LinkedListMultimap.create(), LinkedListMultimap.create(), emptySet(), emptyMap()),
+ emptySet());
+ final UnboundedReadEvaluatorFactory factory = new UnboundedReadEvaluatorFactory(context);
+
+ final Read.Unbounded<String> unbounded = Read.from(source);
+ final Pipeline pipeline = Pipeline.create(options);
+ final PCollection<String> pCollection = pipeline.apply(unbounded);
+ final AppliedPTransform<PBegin, PCollection<String>, Read.Unbounded<String>> application =
+ AppliedPTransform.of(
+ "test", new HashMap<>(), singletonMap(new TupleTag(), pCollection),
+ unbounded, pipeline);
+ final TransformEvaluator<UnboundedSourceShard<String, TestCheckpointMark>> evaluator =
+ factory.forApplication(application, null);
+ final UnboundedSource.UnboundedReader<String> reader = source.createReader(options, null);
+ final UnboundedSourceShard<String, TestCheckpointMark> shard = UnboundedSourceShard.of(
+ source, new NeverDeduplicator(), reader, null);
+ final WindowedValue<UnboundedSourceShard<String, TestCheckpointMark>> value =
+ WindowedValue.of(
+ shard, BoundedWindow.TIMESTAMP_MAX_VALUE, GlobalWindow.INSTANCE, PaneInfo.NO_FIRING);
+ TestUnboundedSource.readerClosedCount = 0;
+ evaluator.processElement(value);
+ }
+
/**
* A terse alias for producing timestamped longs in the {@link GlobalWindow}, where
* the timestamp is the epoch offset by the value of the element.
--
To stop receiving notification emails like this one, please contact
lcwik@apache.org.