You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2024/10/23 14:12:42 UTC
(flink) branch master updated: [FLINK-36591] Remove `flink-contrib`
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 28ae3ae65d1 [FLINK-36591] Remove `flink-contrib`
28ae3ae65d1 is described below
commit 28ae3ae65d13b98dc28b919ef75c14780c512618
Author: Martijn Visser <29...@users.noreply.github.com>
AuthorDate: Wed Oct 23 09:14:23 2024 +0200
[FLINK-36591] Remove `flink-contrib`
This module was introduced with FLINK-1452 to add new contributions in, but only contains a Wikiedits connector that connected to IRC. However, that is now consistently failing and contributions are either merged in directly into OSS Flink (via FLIP/Dev discussion), https://github.com/flink-extended and/or https://flink-packages.org/
---
flink-contrib/README.md | 4 -
flink-contrib/flink-connector-wikiedits/README.md | 38 ----
flink-contrib/flink-connector-wikiedits/pom.xml | 52 -----
.../connectors/wikiedits/WikipediaEditEvent.java | 228 ---------------------
.../legacy/WikipediaEditEventIrcStream.java | 194 ------------------
.../wikiedits/legacy/WikipediaEditsSource.java | 107 ----------
.../wikiedits/WikipediaEditsSourceTest.java | 173 ----------------
.../org.junit.jupiter.api.extension.Extension | 16 --
.../src/test/resources/log4j2-test.properties | 28 ---
flink-contrib/pom.xml | 48 -----
pom.xml | 1 -
tools/ci/stage.sh | 1 -
12 files changed, 890 deletions(-)
diff --git a/flink-contrib/README.md b/flink-contrib/README.md
deleted file mode 100644
index 861d8a19e32..00000000000
--- a/flink-contrib/README.md
+++ /dev/null
@@ -1,4 +0,0 @@
-## Purpose of this module
-
-`flink-contrib` is a staging/incubating area for new modules while the community evaluates
-how they are received by users and whether there will be commitment to maintain them in long term.
diff --git a/flink-contrib/flink-connector-wikiedits/README.md b/flink-contrib/flink-connector-wikiedits/README.md
deleted file mode 100644
index 0e218755f77..00000000000
--- a/flink-contrib/flink-connector-wikiedits/README.md
+++ /dev/null
@@ -1,38 +0,0 @@
-# flink-connector-wikiedits
-
-A non-parallel source that parses a live stream of Wikipedia edits.
-
-Meta data about the edits is mirrored to the IRC channel `#en.wikipedia`. The
-source establishes a connection to this IRC channel and parses the messages
-into `WikipediaEditEvent` instances.
-
-The purpose of this source is to ease the setup of demos of the `DataStream`
-API with live data.
-
-The original idea is from the [Hello Samza](https://samza.apache.org/startup/hello-samza/latest/)
-project of [Apache Samza](http://samza.apache.org). The Samza code for this is located in the
-[samza-hello-samza](https://github.com/apache/samza-hello-samza) repository.
-
-## Example
-
-Add the following dependency to your project:
-
-```xml
-<dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-wikiedits</artifactId>
- <version>1.0-SNAPSHOT</version>
-</dependency>
-```
-
-You can use the source like regular sources:
-
-```java
-StreamExecutionEnvironment env = StreamExecutionEnvironment
- .getExecutionEnvironment();
-
-DataStream<WikipediaEditEvent> edits = env
- .addSource(new WikipediaEditsSource());
-```
-Remember that it is *non-parallel* source and as such it will run with
-parallelism 1.
diff --git a/flink-contrib/flink-connector-wikiedits/pom.xml b/flink-contrib/flink-connector-wikiedits/pom.xml
deleted file mode 100644
index 7659220bfb8..00000000000
--- a/flink-contrib/flink-connector-wikiedits/pom.xml
+++ /dev/null
@@ -1,52 +0,0 @@
-<?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/maven-v4_0_0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-contrib</artifactId>
- <version>2.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>flink-connector-wikiedits</artifactId>
- <name>Flink : Contrib : Connectors : Wikiedits</name>
-
- <packaging>jar</packaging>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</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>
- </dependency>
- </dependencies>
-</project>
diff --git a/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditEvent.java b/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditEvent.java
deleted file mode 100644
index 770e244bbdd..00000000000
--- a/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditEvent.java
+++ /dev/null
@@ -1,228 +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.flink.streaming.connectors.wikiedits;
-
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-/** Instances of this class represent edits made on Wikipedia. */
-public class WikipediaEditEvent {
-
- // Metadata
- private final long timestamp;
- private final String channel;
-
- // Edit attributes
- private final String title;
- private final String diffUrl;
- private final String user;
- private final int byteDiff;
- private final String summary;
- private final int flags;
-
- public WikipediaEditEvent(
- long timestamp,
- String channel,
- String title,
- String diffUrl,
- String user,
- int byteDiff,
- String summary,
- boolean isMinor,
- boolean isNew,
- boolean isUnpatrolled,
- boolean isBotEdit,
- boolean isSpecial,
- boolean isTalk) {
-
- if (channel == null
- || title == null
- || diffUrl == null
- || user == null
- || summary == null) {
- throw new NullPointerException();
- }
-
- this.timestamp = timestamp;
- this.channel = channel;
-
- this.title = title;
- this.diffUrl = diffUrl;
- this.user = user;
- this.byteDiff = byteDiff;
- this.summary = summary;
- this.flags = getFlags(isMinor, isNew, isUnpatrolled, isBotEdit, isSpecial, isTalk);
- }
-
- /**
- * Returns the timestamp when this event arrived at the source.
- *
- * @return The timestamp assigned at the source.
- */
- public long getTimestamp() {
- return timestamp;
- }
-
- public String getChannel() {
- return channel;
- }
-
- public String getTitle() {
- return title;
- }
-
- public String getDiffUrl() {
- return diffUrl;
- }
-
- public String getUser() {
- return user;
- }
-
- public int getByteDiff() {
- return byteDiff;
- }
-
- public String getSummary() {
- return summary;
- }
-
- public boolean isMinor() {
- return (flags & IS_MINOR) > 0;
- }
-
- public boolean isNew() {
- return (flags & IS_NEW) > 0;
- }
-
- public boolean isUnpatrolled() {
- return (flags & IS_UNPATROLLED) > 0;
- }
-
- public boolean isBotEdit() {
- return (flags & IS_BOT_EDIT) > 0;
- }
-
- public boolean isSpecial() {
- return (flags & IS_SPECIAL) > 0;
- }
-
- public boolean isTalk() {
- return (flags & IS_TALK) > 0;
- }
-
- @Override
- public String toString() {
- return "WikipediaEditEvent{"
- + "timestamp="
- + timestamp
- + ", channel='"
- + channel
- + '\''
- + ", title='"
- + title
- + '\''
- + ", diffUrl='"
- + diffUrl
- + '\''
- + ", user='"
- + user
- + '\''
- + ", byteDiff="
- + byteDiff
- + ", summary='"
- + summary
- + '\''
- + ", flags="
- + flags
- + '}';
- }
-
- // - Flags ----------------------------------------------------------------
-
- private static final byte IS_MINOR = 0B000001;
- private static final byte IS_NEW = 0B000010;
- private static final byte IS_UNPATROLLED = 0B000100;
- private static final byte IS_BOT_EDIT = 0B001000;
- private static final byte IS_SPECIAL = 0B010000;
- private static final byte IS_TALK = 0B100000;
-
- private byte getFlags(
- boolean isMinor,
- boolean isNew,
- boolean isUnpatrolled,
- boolean isBotEdit,
- boolean isSpecial,
- boolean isTalk) {
-
- byte flag = 0;
- flag |= isMinor ? IS_MINOR : flag;
- flag |= isNew ? IS_NEW : flag;
- flag |= isUnpatrolled ? IS_UNPATROLLED : flag;
- flag |= isBotEdit ? IS_BOT_EDIT : flag;
- flag |= isSpecial ? IS_SPECIAL : flag;
- flag |= isTalk ? IS_TALK : flag;
-
- return flag;
- }
-
- // - Parser ---------------------------------------------------------------
-
- /** Expected pattern of raw events. */
- private static final Pattern p =
- Pattern.compile(
- "\\[\\[(.*)\\]\\]\\s(.*)\\s(.*)\\s\\*\\s(.*)\\s\\*\\s\\(\\+?(.\\d*)\\)\\s(.*)");
-
- public static WikipediaEditEvent fromRawEvent(long timestamp, String channel, String rawEvent) {
- final Matcher m = p.matcher(rawEvent);
-
- if (m.find() && m.groupCount() == 6) {
- String title = m.group(1);
- String flags = m.group(2);
- String diffUrl = m.group(3);
- String user = m.group(4);
- int byteDiff = Integer.parseInt(m.group(5));
- String summary = m.group(6);
-
- boolean isMinor = flags.contains("M");
- boolean isNew = flags.contains("N");
- boolean isUnpatrolled = flags.contains("!");
- boolean isBotEdit = flags.contains("B");
- boolean isSpecial = title.startsWith("Special:");
- boolean isTalk = title.startsWith("Talk:");
-
- return new WikipediaEditEvent(
- timestamp,
- channel,
- title,
- diffUrl,
- user,
- byteDiff,
- summary,
- isMinor,
- isNew,
- isUnpatrolled,
- isBotEdit,
- isSpecial,
- isTalk);
- }
-
- return null;
- }
-}
diff --git a/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/legacy/WikipediaEditEventIrcStream.java b/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/legacy/WikipediaEditEventIrcStream.java
deleted file mode 100644
index 00a7de0318c..00000000000
--- a/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/legacy/WikipediaEditEventIrcStream.java
+++ /dev/null
@@ -1,194 +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.flink.streaming.connectors.wikiedits.legacy;
-
-import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditEvent;
-
-import org.schwering.irc.lib.IRCConnection;
-import org.schwering.irc.lib.IRCEventListener;
-import org.schwering.irc.lib.IRCModeParser;
-import org.schwering.irc.lib.IRCUser;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.Objects;
-import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-
-class WikipediaEditEventIrcStream implements AutoCloseable {
-
- private static final Logger LOG = LoggerFactory.getLogger(WikipediaEditEventIrcStream.class);
-
- /** Bounded queue of edit events from the channel. */
- private final BlockingQueue<WikipediaEditEvent> edits = new ArrayBlockingQueue<>(128);
-
- /** IRC connection (NOTE: this is a separate Thread). */
- private IRCConnection conn;
-
- WikipediaEditEventIrcStream(String host, int port) {
- final String nick = "flink-bot-" + UUID.randomUUID().toString();
- this.conn = new IRCConnection(host, new int[] {port}, "", nick, nick, nick);
- conn.addIRCEventListener(new WikipediaIrcChannelListener(edits));
- conn.setEncoding("UTF-8");
- conn.setPong(true);
- conn.setColors(false);
- conn.setDaemon(true);
- conn.setName("WikipediaEditEventIrcStreamThread");
- }
-
- BlockingQueue<WikipediaEditEvent> getEdits() {
- return edits;
- }
-
- void connect() throws IOException {
- if (!conn.isConnected()) {
- conn.connect();
- }
- }
-
- void join(String channel) {
- Objects.requireNonNull(channel, "channel");
- conn.send("JOIN " + channel);
- }
-
- void leave(String channel) {
- conn.send("PART " + channel);
- }
-
- @Override
- public void close() throws Exception {
- if (conn != null && conn.isConnected()) {
- conn.doQuit();
- conn.close();
- conn.join(5 * 1000);
- }
- }
-
- // ------------------------------------------------------------------------
- // IRC channel listener
- // ------------------------------------------------------------------------
-
- private static class WikipediaIrcChannelListener implements IRCEventListener {
-
- private final BlockingQueue<WikipediaEditEvent> edits;
-
- WikipediaIrcChannelListener(BlockingQueue<WikipediaEditEvent> edits) {
- this.edits = Objects.requireNonNull(edits, "edits");
- }
-
- @Override
- public void onPrivmsg(String target, IRCUser user, String msg) {
- LOG.debug("[{}] {}: {}.", target, user.getNick(), msg);
-
- WikipediaEditEvent event =
- WikipediaEditEvent.fromRawEvent(System.currentTimeMillis(), target, msg);
-
- if (event != null) {
- if (!edits.offer(event)) {
- LOG.debug("Dropping message, because of full queue.");
- }
- }
- }
-
- @Override
- public void onRegistered() {
- LOG.debug("Connected.");
- }
-
- @Override
- public void onDisconnected() {
- LOG.debug("Disconnected.");
- }
-
- @Override
- public void onError(String msg) {
- LOG.error("Error: '{}'.", msg);
- }
-
- @Override
- public void onError(int num, String msg) {
- LOG.error("Error #{}: '{}'.", num, msg);
- }
-
- @Override
- public void onInvite(String chan, IRCUser user, String passiveNick) {
- LOG.debug("[{}]: {} invites {}.", chan, user.getNick(), passiveNick);
- }
-
- @Override
- public void onJoin(String chan, IRCUser user) {
- LOG.debug("[{}]: {} joins.", chan, user.getNick());
- }
-
- @Override
- public void onKick(String chan, IRCUser user, String passiveNick, String msg) {
- LOG.debug("[{}]: {} kicks {}.", chan, user.getNick(), passiveNick);
- }
-
- @Override
- public void onMode(String chan, IRCUser user, IRCModeParser modeParser) {
- LOG.debug("[{}]: mode '{}'.", chan, modeParser.getLine());
- }
-
- @Override
- public void onMode(IRCUser user, String passiveNick, String mode) {
- LOG.debug("{} sets modes {} ({}).", user.getNick(), mode, passiveNick);
- }
-
- @Override
- public void onNick(IRCUser user, String newNick) {
- LOG.debug("{} is now known as {}.", user.getNick(), newNick);
- }
-
- @Override
- public void onNotice(String target, IRCUser user, String msg) {
- LOG.debug("[{}] {} (notice): {}.", target, user.getNick(), msg);
- }
-
- @Override
- public void onPart(String chan, IRCUser user, String msg) {
- LOG.debug("[{}] {} parts {}.", chan, user.getNick(), msg);
- }
-
- @Override
- public void onPing(String ping) {}
-
- @Override
- public void onQuit(IRCUser user, String msg) {
- LOG.debug("Quit: {}.", user.getNick());
- }
-
- @Override
- public void onReply(int num, String value, String msg) {
- LOG.debug("Reply #{}: {} {}.", num, value, msg);
- }
-
- @Override
- public void onTopic(String chan, IRCUser user, String topic) {
- LOG.debug("[{}] {} changes topic into {}.", chan, user.getNick(), topic);
- }
-
- @Override
- public void unknown(String prefix, String command, String middle, String trailing) {
- LOG.warn("UNKNOWN: " + prefix + " " + command + " " + middle + " " + trailing);
- }
- }
-}
diff --git a/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/legacy/WikipediaEditsSource.java b/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/legacy/WikipediaEditsSource.java
deleted file mode 100644
index 9c905450385..00000000000
--- a/flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/legacy/WikipediaEditsSource.java
+++ /dev/null
@@ -1,107 +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.flink.streaming.connectors.wikiedits.legacy;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.streaming.api.functions.source.legacy.RichSourceFunction;
-import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
-import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditEvent;
-
-import java.util.Objects;
-import java.util.concurrent.TimeUnit;
-
-/**
- * This class is a SourceFunction that reads {@link WikipediaEditEvent} instances from the IRC
- * channel <code>#en.wikipedia</code>.
- *
- * @deprecated This class is based on the {@link SourceFunction} API, which is due to be removed.
- * Use the new {@link org.apache.flink.api.connector.source.Source} API instead.
- */
-@Internal
-public class WikipediaEditsSource extends RichSourceFunction<WikipediaEditEvent> {
-
- /** Hostname of the server to connect to. */
- public static final String DEFAULT_HOST = "irc.wikimedia.org";
-
- /** Port of the server to connect to. */
- public static final int DEFAULT_PORT = 6667;
-
- /** IRC channel to join. */
- public static final String DEFAULT_CHANNEL = "#en.wikipedia";
-
- private final String host;
- private final int port;
- private final String channel;
-
- private volatile boolean isRunning = true;
-
- /**
- * Creates a source reading {@link WikipediaEditEvent} instances from the IRC channel <code>
- * #en.wikipedia</code>.
- *
- * <p>This creates a separate Thread for the IRC connection.
- */
- public WikipediaEditsSource() {
- this(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_CHANNEL);
- }
-
- /**
- * Creates a source reading {@link WikipediaEditEvent} instances from the specified IRC channel.
- *
- * <p>In most cases, you want to use the default {@link #WikipediaEditsSource} constructor. This
- * constructor is meant to be used only if there is a problem with the default constructor.
- *
- * @param host The IRC server to connect to.
- * @param port The port of the IRC server to connect to.
- * @param channel The channel to join. Messages not matching the expected format will be
- * ignored.
- */
- public WikipediaEditsSource(String host, int port, String channel) {
- this.host = host;
- this.port = port;
- this.channel = Objects.requireNonNull(channel);
- }
-
- @Override
- public void run(SourceContext<WikipediaEditEvent> ctx) throws Exception {
- try (WikipediaEditEventIrcStream ircStream = new WikipediaEditEventIrcStream(host, port)) {
- // Open connection and join channel
- ircStream.connect();
- ircStream.join(channel);
-
- try {
- while (isRunning) {
- // Query for the next edit event
- WikipediaEditEvent edit = ircStream.getEdits().poll(100, TimeUnit.MILLISECONDS);
-
- if (edit != null) {
- ctx.collect(edit);
- }
- }
- } finally {
- ircStream.leave(channel);
- }
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
-}
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
deleted file mode 100644
index 6a4eeb671da..00000000000
--- a/flink-contrib/flink-connector-wikiedits/src/test/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditsSourceTest.java
+++ /dev/null
@@ -1,173 +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.flink.streaming.connectors.wikiedits;
-
-import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.connectors.wikiedits.legacy.WikipediaEditsSource;
-import org.apache.flink.testutils.junit.RetryOnFailure;
-import org.apache.flink.testutils.junit.extensions.retry.RetryExtension;
-
-import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.time.Duration;
-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.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
-
-/** Tests for the WikipediaEditsSource. */
-@ExtendWith(RetryExtension.class)
-class WikipediaEditsSourceTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(WikipediaEditsSourceTest.class);
-
- /** We first check the connection to the IRC server. If it fails, this test is ignored. */
- @TestTemplate
- @RetryOnFailure(times = 1)
- void testWikipediaEditsSource() throws Exception {
- if (canConnect(1, TimeUnit.SECONDS)) {
- final Duration testTimeout = Duration.ofSeconds(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 something 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);
-
- Object event = null;
- Exception error = null;
-
- // 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());
- }
-
- assertThat(event)
- .as("Did not receive a WikipediaEditEvent within the desired timeout")
- .isNotNull();
- assertThat(event)
- .as("Received unexpected event " + event)
- .isInstanceOf(WikipediaEditEvent.class);
- } finally {
- wikipediaEditsSource.cancel();
-
- if (executorService != null) {
- executorService.shutdownNow();
- executorService.awaitTermination(1, TimeUnit.SECONDS);
- }
- }
- } else {
- LOG.info("Skipping test, because not able to connect to IRC server.");
- }
- }
-
- private long deadlineNanos(Duration testTimeout) {
- return System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(testTimeout.toMillis());
- }
-
- private static class CollectingSourceContext<T> implements SourceFunction.SourceContext<T> {
-
- private final BlockingQueue<Object> events;
-
- private CollectingSourceContext(BlockingQueue<Object> events) {
- this.events = Objects.requireNonNull(events, "events");
- }
-
- @Override
- public void collect(T element) {
- events.offer(element);
- }
-
- @Override
- public void collectWithTimestamp(T element, long timestamp) {
- throw new UnsupportedOperationException();
- }
-
- @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;
- }
-}
diff --git a/flink-contrib/flink-connector-wikiedits/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/flink-contrib/flink-connector-wikiedits/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
deleted file mode 100644
index 28999133c2b..00000000000
--- a/flink-contrib/flink-connector-wikiedits/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
+++ /dev/null
@@ -1,16 +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.
-
-org.apache.flink.util.TestLoggerExtension
\ No newline at end of file
diff --git a/flink-contrib/flink-connector-wikiedits/src/test/resources/log4j2-test.properties b/flink-contrib/flink-connector-wikiedits/src/test/resources/log4j2-test.properties
deleted file mode 100644
index 835c2ec9a3d..00000000000
--- a/flink-contrib/flink-connector-wikiedits/src/test/resources/log4j2-test.properties
+++ /dev/null
@@ -1,28 +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.
-################################################################################
-
-# Set root logger level to OFF to not flood build logs
-# set manually to INFO for debugging purposes
-rootLogger.level = OFF
-rootLogger.appenderRef.test.ref = TestLogger
-
-appender.testlogger.name = TestLogger
-appender.testlogger.type = CONSOLE
-appender.testlogger.target = SYSTEM_ERR
-appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-contrib/pom.xml b/flink-contrib/pom.xml
deleted file mode 100644
index 1e1e051ebf8..00000000000
--- a/flink-contrib/pom.xml
+++ /dev/null
@@ -1,48 +0,0 @@
-<?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/maven-v4_0_0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-parent</artifactId>
- <version>2.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>flink-contrib</artifactId>
- <name>Flink : Contrib : </name>
-
- <packaging>pom</packaging>
-
- <modules>
- <module>flink-connector-wikiedits</module>
- </modules>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-test-utils-junit</artifactId>
- </dependency>
- </dependencies>
-</project>
diff --git a/pom.xml b/pom.xml
index ee89fba30b0..60fb82726e2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -86,7 +86,6 @@ under the License.
<module>flink-libraries</module>
<module>flink-table</module>
<module>flink-quickstart</module>
- <module>flink-contrib</module>
<module>flink-dist</module>
<module>flink-dist-scala</module>
<module>flink-metrics</module>
diff --git a/tools/ci/stage.sh b/tools/ci/stage.sh
index 1f4801a3ede..4e4e20a4a97 100755
--- a/tools/ci/stage.sh
+++ b/tools/ci/stage.sh
@@ -80,7 +80,6 @@ flink-table/flink-table-test-utils,\
"
MODULES_CONNECTORS="\
-flink-contrib/flink-connector-wikiedits,\
flink-filesystems,\
flink-filesystems/flink-azure-fs-hadoop,\
flink-filesystems/flink-fs-hadoop-shaded,\