flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ches...@apache.org
Subject [2/2] flink git commit: [FLINK-7762] [connector-wikiedits] Make WikipediaEditsSourceTest proper test
Date Mon, 04 Dec 2017 13:29:28 GMT
[FLINK-7762] [connector-wikiedits] Make WikipediaEditsSourceTest proper test

The WikipediaEditsSourceTest unnecessarily implements an integration
test that starts a FlinkMiniCluster and executes a small Flink program.

This simply creates a source and executes run in a separate thread until
a single WikipediaEditEvent is received.

This closes #5102.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e7ca1055
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e7ca1055
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e7ca1055

Branch: refs/heads/master
Commit: e7ca1055afc46e4a6ccb3492758de36bfd6419a3
Parents: ec0c416
Author: Ufuk Celebi <uce@apache.org>
Authored: Wed Nov 29 16:36:29 2017 +0100
Committer: zentol <chesnay@apache.org>
Committed: Mon Dec 4 14:29:01 2017 +0100

----------------------------------------------------------------------
 flink-contrib/flink-connector-wikiedits/pom.xml |   7 +-
 .../wikiedits/WikipediaEditsSourceTest.java     | 172 +++++++++++++------
 2 files changed, 125 insertions(+), 54 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/e7ca1055/flink-contrib/flink-connector-wikiedits/pom.xml
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-connector-wikiedits/pom.xml b/flink-contrib/flink-connector-wikiedits/pom.xml
index c13f93e..3908658 100644
--- a/flink-contrib/flink-connector-wikiedits/pom.xml
+++ b/flink-contrib/flink-connector-wikiedits/pom.xml
@@ -40,11 +40,16 @@ under the License.
 			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
 			<version>${project.version}</version>
 		</dependency>
-
 		<dependency>
 			<groupId>org.schwering</groupId>
 			<artifactId>irclib</artifactId>
 			<version>1.10</version>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-test-utils-junit</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 </project>

http://git-wip-us.apache.org/repos/asf/flink/blob/e7ca1055/flink-contrib/flink-connector-wikiedits/src/test/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditsSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-connector-wikiedits/src/test/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditsSourceTest.java
b/flink-contrib/flink-connector-wikiedits/src/test/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditsSourceTest.java
index 439aa36..f6fa8e0 100644
--- a/flink-contrib/flink-connector-wikiedits/src/test/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditsSourceTest.java
+++ b/flink-contrib/flink-connector-wikiedits/src/test/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditsSourceTest.java
@@ -18,19 +18,29 @@
 
 package org.apache.flink.streaming.connectors.wikiedits;
 
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.testutils.junit.RetryOnFailure;
+import org.apache.flink.testutils.junit.RetryRule;
 
+import org.junit.Rule;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 /**
@@ -40,71 +50,127 @@ public class WikipediaEditsSourceTest {
 
 	private static final Logger LOG = LoggerFactory.getLogger(WikipediaEditsSourceTest.class);
 
+	@Rule
+	public RetryRule retryRule = new RetryRule();
+
 	/**
-	 * NOTE: if you are behind a firewall you may need to use a SOCKS Proxy for this test.
-	 *
-	 * <p>We first check the connection to the IRC server. If it fails, this test
-	 * is effectively ignored.
-	 *
-	 * @see <a href="http://docs.oracle.com/javase/8/docs/technotes/guides/net/proxies.html">Socks
Proxy</a>
+	 * We first check the connection to the IRC server. If it fails, this test is ignored.
 	 */
-	@Test(timeout = 120 * 1000)
+	@Test
+	@RetryOnFailure(times = 1)
 	public void testWikipediaEditsSource() throws Exception {
-		final int numRetries = 5;
-		final int waitBetweenRetriesMillis = 2000;
-		final int connectTimeout = 1000;
+		if (canConnect(1, TimeUnit.SECONDS)) {
+			final Time testTimeout = Time.seconds(60);
+			final WikipediaEditsSource wikipediaEditsSource = new WikipediaEditsSource();
+
+			ExecutorService executorService = null;
+			try {
+				executorService = Executors.newSingleThreadExecutor();
+				BlockingQueue<Object> collectedEvents = new ArrayBlockingQueue<>(1);
+				AtomicReference<Exception> asyncError = new AtomicReference<>();
+
+				// Execute the source in a different thread and collect events into the queue.
+				// We do this in a separate thread in order to not block the main test thread
+				// indefinitely in case that somethign bad happens (like not receiving any
+				// events)
+				executorService.execute(() -> {
+					try {
+						wikipediaEditsSource.run(new CollectingSourceContext<>(collectedEvents));
+					} catch (Exception e) {
+						boolean interrupted = e instanceof InterruptedException;
+						if (!interrupted) {
+							LOG.warn("Failure in WikipediaEditsSource", e);
+						}
+
+						asyncError.compareAndSet(null, e);
+					}
+				});
+
+				long deadline = deadlineNanos(testTimeout);
 
-		boolean success = false;
+				Object event = null;
+				Exception error = null;
 
-		for (int i = 0; i < numRetries && !success; i++) {
-			// Check connection
-			boolean canConnect = false;
+				// Check event or error
+				while (event == null && error == null && System.nanoTime() < deadline)
{
+					event = collectedEvents.poll(1, TimeUnit.SECONDS);
+					error = asyncError.get();
+				}
+
+				if (error != null) {
+					// We don't use assertNull, because we want to include the error message
+					fail("Failure in WikipediaEditsSource: " + error.getMessage());
+				}
 
-			String host = WikipediaEditsSource.DEFAULT_HOST;
-			int port = WikipediaEditsSource.DEFAULT_PORT;
+				assertNotNull("Did not receive a WikipediaEditEvent within the desired timeout", event);
+				assertTrue("Received unexpected event " + event, event instanceof WikipediaEditEvent);
+			} finally {
+				wikipediaEditsSource.cancel();
 
-			try (Socket s = new Socket()) {
-				s.connect(new InetSocketAddress(host, port), connectTimeout);
-				canConnect = s.isConnected();
-			} catch (Throwable ignored) {
+				if (executorService != null) {
+					executorService.shutdownNow();
+					executorService.awaitTermination(1, TimeUnit.SECONDS);
+				}
 			}
+		} else {
+			LOG.info("Skipping test, because not able to connect to IRC server.");
+		}
+	}
 
-			if (canConnect) {
-				StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-				env.getConfig().disableSysoutLogging();
+	private long deadlineNanos(Time testTimeout) {
+		return System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(testTimeout.toMilliseconds());
+	}
 
-				DataStream<WikipediaEditEvent> edits = env.addSource(new WikipediaEditsSource());
+	private static class CollectingSourceContext<T> implements SourceFunction.SourceContext<T>
{
 
-				edits.addSink(new SinkFunction<WikipediaEditEvent>() {
-					@Override
-					public void invoke(WikipediaEditEvent value) throws Exception {
-						throw new Exception("Expected test exception");
-					}
-				});
+		private final BlockingQueue<Object> events;
 
-				try {
-					env.execute();
-					fail("Did not throw expected Exception.");
-				} catch (Exception e) {
-					assertNotNull(e.getCause());
-					assertEquals("Expected test exception", e.getCause().getMessage());
-				}
+		private CollectingSourceContext(BlockingQueue<Object> events) {
+			this.events = Objects.requireNonNull(events, "events");
+		}
 
-				success = true;
-			} else {
-				LOG.info("Failed to connect to IRC server ({}/{}). Retrying in {} ms.",
-						i + 1,
-						numRetries,
-						waitBetweenRetriesMillis);
+		@Override
+		public void collect(T element) {
+			events.offer(element);
+		}
+
+		@Override
+		public void collectWithTimestamp(T element, long timestamp) {
+			throw new UnsupportedOperationException();
 
-				Thread.sleep(waitBetweenRetriesMillis);
-			}
 		}
 
-		if (success) {
-			LOG.info("Successfully ran test.");
-		} else {
-			LOG.info("Skipped test, because not able to connect to IRC server.");
+		@Override
+		public void emitWatermark(Watermark mark) {
+			throw new UnsupportedOperationException();
+		}
+
+		@Override
+		public void markAsTemporarilyIdle() {
+			throw new UnsupportedOperationException();
+		}
+
+		@Override
+		public Object getCheckpointLock() {
+			throw new UnsupportedOperationException();
 		}
+
+		@Override
+		public void close() {
+		}
+
+	}
+
+	private static boolean canConnect(int timeout, TimeUnit unit) {
+		String host = WikipediaEditsSource.DEFAULT_HOST;
+		int port = WikipediaEditsSource.DEFAULT_PORT;
+
+		try (Socket s = new Socket()) {
+			s.connect(new InetSocketAddress(host, port), (int) unit.toMillis(timeout));
+			return s.isConnected();
+		} catch (Throwable ignored) {
+		}
+
+		return false;
 	}
 }


Mime
View raw message