You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by da...@apache.org on 2023/08/04 12:25:46 UTC
[flink-connector-mongodb] branch v1.0 updated (bea5957 -> 723b34a)
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a change to branch v1.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-mongodb.git
from bea5957 [FLINK-32024][docs] Short code related to externalized connector retrieve version from its own data yaml
new 23b3789 [FLINK-32187] Remove dependency on flink-shaded
new 99755b4 [FLINK-31923] Run nightly builds against multiple branches and Flink versions
new a351ddf [hotfix] Test PRs against 1.17-SNAPSHOT and 1.18-SNAPSHOT
new 1de4999 [FLINK-32446][connectors/mongodb] MongoWriter should regularly check whether the latest flush time is arriving
new 723b34a [FLINK-32348][connectors/mongodb][tests] Fix MongoDB tests are flaky and time out
The 5 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:
.editorconfig | 2 +-
.github/workflows/push_pr.yml | 5 +-
.github/workflows/weekly.yml | 24 ++++++-
.../connector/mongodb/sink/writer/MongoWriter.java | 83 +++++++++++++++++++---
.../source/enumerator/MongoSourceEnumerator.java | 13 ++--
.../reader/split/MongoScanSourceSplitReader.java | 1 -
.../mongodb/sink/writer/MongoWriterITCase.java | 7 +-
.../connector/mongodb/testutils/MongoTestUtil.java | 13 ++++
pom.xml | 12 ----
tools/maven/checkstyle.xml | 2 +-
10 files changed, 129 insertions(+), 33 deletions(-)
[flink-connector-mongodb] 01/05: [FLINK-32187] Remove dependency on flink-shaded
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch v1.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-mongodb.git
commit 23b37894e2bdd82f4956533e331f3aab22576694
Author: Sergey Nuyanzin <se...@aiven.io>
AuthorDate: Thu May 25 12:06:10 2023 +0200
[FLINK-32187] Remove dependency on flink-shaded
---
.editorconfig | 2 +-
pom.xml | 12 ------------
tools/maven/checkstyle.xml | 2 +-
3 files changed, 2 insertions(+), 14 deletions(-)
diff --git a/.editorconfig b/.editorconfig
index b66771d..ab5d802 100644
--- a/.editorconfig
+++ b/.editorconfig
@@ -113,7 +113,7 @@ ij_java_doc_preserve_line_breaks = false
# ij_java_generate_final_locals = false
# ij_java_generate_final_parameters = false
# ij_java_if_brace_force = never
-ij_java_imports_layout = org.apache.flink.**,|,org.apache.flink.shaded.**,|,*,|,javax.**,|,java.**,|,scala.**,|,$*
+ij_java_imports_layout = org.apache.flink.**,|,|,*,|,javax.**,|,java.**,|,scala.**,|,$*
# ij_java_indent_case_from_switch = true
# ij_java_insert_inner_class_imports = false
# ij_java_insert_override_annotation = true
diff --git a/pom.xml b/pom.xml
index 65ff786..8389f9b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,7 +54,6 @@ under the License.
<mongodb.version>4.7.2</mongodb.version>
<flink.version>1.17.0</flink.version>
- <flink.shaded.version>16.1</flink.shaded.version>
<junit5.version>5.8.1</junit5.version>
<assertj.version>3.21.0</assertj.version>
@@ -77,11 +76,6 @@ under the License.
</modules>
<dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-force-shading</artifactId>
- </dependency>
-
<!-- Root dependencies for all projects -->
<!-- Logging API -->
@@ -194,12 +188,6 @@ under the License.
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-force-shading</artifactId>
- <version>${flink.shaded.version}</version>
- </dependency>
-
<!-- Flink ArchUnit -->
<dependency>
diff --git a/tools/maven/checkstyle.xml b/tools/maven/checkstyle.xml
index 3fc4711..eb62611 100644
--- a/tools/maven/checkstyle.xml
+++ b/tools/maven/checkstyle.xml
@@ -195,7 +195,7 @@ This file is based on the checkstyle file of Apache Beam.
<!-- Checks for out of order import statements. -->
<property name="severity" value="error"/>
<property name="groups"
- value="org.apache.flink,org.apache.flink.shaded,*,javax,java,scala"/>
+ value="org.apache.flink,*,javax,java,scala"/>
<property name="separated" value="true"/>
<property name="sortStaticImportsAlphabetically" value="true"/>
<property name="option" value="bottom"/>
[flink-connector-mongodb] 03/05: [hotfix] Test PRs against 1.17-SNAPSHOT and 1.18-SNAPSHOT
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch v1.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-mongodb.git
commit a351ddf63e2e04aae403d164bc2faab5c5fff767
Author: Martijn Visser <mv...@confluent.io>
AuthorDate: Thu Jun 15 10:54:56 2023 +0200
[hotfix] Test PRs against 1.17-SNAPSHOT and 1.18-SNAPSHOT
---
.github/workflows/push_pr.yml | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
index 335f108..8cc815d 100644
--- a/.github/workflows/push_pr.yml
+++ b/.github/workflows/push_pr.yml
@@ -23,6 +23,9 @@ concurrency:
cancel-in-progress: true
jobs:
compile_and_test:
+ strategy:
+ matrix:
+ flink: [1.17-SNAPSHOT, 1.18-SNAPSHOT]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
- flink_version: 1.17.0
+ flink_version: ${{ matrix.flink }}
[flink-connector-mongodb] 04/05: [FLINK-32446][connectors/mongodb] MongoWriter should regularly check whether the latest flush time is arriving
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch v1.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-mongodb.git
commit 1de499900fdb4d934d95c52808df6516d627ed58
Author: Jiabao Sun <ji...@xtransfer.cn>
AuthorDate: Wed Jun 28 12:02:44 2023 +0800
[FLINK-32446][connectors/mongodb] MongoWriter should regularly check whether the latest flush time is arriving
This closes #12.
---
.../connector/mongodb/sink/writer/MongoWriter.java | 83 +++++++++++++++++++---
.../mongodb/sink/writer/MongoWriterITCase.java | 7 +-
.../connector/mongodb/testutils/MongoTestUtil.java | 13 ++++
3 files changed, 92 insertions(+), 11 deletions(-)
diff --git a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriter.java b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriter.java
index df76891..5319959 100644
--- a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriter.java
+++ b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriter.java
@@ -33,6 +33,7 @@ import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.apache.flink.util.Collector;
import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import com.mongodb.MongoException;
import com.mongodb.client.MongoClient;
@@ -45,6 +46,10 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -68,11 +73,18 @@ public class MongoWriter<IN> implements SinkWriter<IN> {
private final Collector<WriteModel<BsonDocument>> collector;
private final Counter numRecordsOut;
private final MongoClient mongoClient;
+ private final long batchIntervalMs;
+ private final int batchSize;
private boolean checkpointInProgress = false;
private volatile long lastSendTime = 0L;
private volatile long ackTime = Long.MAX_VALUE;
+ private transient volatile boolean closed = false;
+ private transient ScheduledExecutorService scheduler;
+ private transient ScheduledFuture<?> scheduledFuture;
+ private transient volatile Exception flushException;
+
public MongoWriter(
MongoConnectionOptions connectionOptions,
MongoWriteOptions writeOptions,
@@ -83,6 +95,8 @@ public class MongoWriter<IN> implements SinkWriter<IN> {
this.writeOptions = checkNotNull(writeOptions);
this.serializationSchema = checkNotNull(serializationSchema);
this.flushOnCheckpoint = flushOnCheckpoint;
+ this.batchIntervalMs = writeOptions.getBatchIntervalMs();
+ this.batchSize = writeOptions.getBatchSize();
checkNotNull(initContext);
this.mailboxExecutor = checkNotNull(initContext.getMailboxExecutor());
@@ -105,10 +119,37 @@ public class MongoWriter<IN> implements SinkWriter<IN> {
// Initialize the mongo client.
this.mongoClient = MongoClients.create(connectionOptions.getUri());
+
+ boolean flushOnlyOnCheckpoint = batchIntervalMs == -1 && batchSize == -1;
+
+ if (!flushOnlyOnCheckpoint && batchIntervalMs > 0) {
+ this.scheduler =
+ Executors.newScheduledThreadPool(1, new ExecutorThreadFactory("mongo-writer"));
+
+ this.scheduledFuture =
+ this.scheduler.scheduleWithFixedDelay(
+ () -> {
+ synchronized (MongoWriter.this) {
+ if (!closed && isOverMaxBatchIntervalLimit()) {
+ try {
+ doBulkWrite();
+ } catch (Exception e) {
+ flushException = e;
+ }
+ }
+ }
+ },
+ batchIntervalMs,
+ batchIntervalMs,
+ TimeUnit.MILLISECONDS);
+ }
}
@Override
- public void write(IN element, Context context) throws IOException, InterruptedException {
+ public synchronized void write(IN element, Context context)
+ throws IOException, InterruptedException {
+ checkFlushException();
+
// do not allow new bulk writes until all actions are flushed
while (checkpointInProgress) {
mailboxExecutor.yield();
@@ -122,7 +163,9 @@ public class MongoWriter<IN> implements SinkWriter<IN> {
}
@Override
- public void flush(boolean endOfInput) throws IOException {
+ public synchronized void flush(boolean endOfInput) throws IOException {
+ checkFlushException();
+
checkpointInProgress = true;
while (!bulkRequests.isEmpty() && (flushOnCheckpoint || endOfInput)) {
doBulkWrite();
@@ -131,8 +174,28 @@ public class MongoWriter<IN> implements SinkWriter<IN> {
}
@Override
- public void close() {
- mongoClient.close();
+ public synchronized void close() throws Exception {
+ if (!closed) {
+ if (scheduledFuture != null) {
+ scheduledFuture.cancel(false);
+ scheduler.shutdown();
+ }
+
+ if (!bulkRequests.isEmpty()) {
+ try {
+ doBulkWrite();
+ } catch (Exception e) {
+ LOG.error("Writing records to MongoDB failed when closing MongoWriter", e);
+ throw new IOException("Writing records to MongoDB failed.", e);
+ } finally {
+ mongoClient.close();
+ closed = true;
+ }
+ } else {
+ mongoClient.close();
+ closed = true;
+ }
+ }
}
@VisibleForTesting
@@ -172,13 +235,17 @@ public class MongoWriter<IN> implements SinkWriter<IN> {
}
private boolean isOverMaxBatchSizeLimit() {
- int bulkActions = writeOptions.getBatchSize();
- return bulkActions != -1 && bulkRequests.size() >= bulkActions;
+ return batchSize != -1 && bulkRequests.size() >= batchSize;
}
private boolean isOverMaxBatchIntervalLimit() {
- long bulkFlushInterval = writeOptions.getBatchIntervalMs();
long lastSentInterval = System.currentTimeMillis() - lastSendTime;
- return bulkFlushInterval != -1 && lastSentInterval >= bulkFlushInterval;
+ return batchIntervalMs != -1 && lastSentInterval >= batchIntervalMs;
+ }
+
+ private void checkFlushException() {
+ if (flushException != null) {
+ throw new RuntimeException("Writing records to MongoDB failed.", flushException);
+ }
}
}
diff --git a/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriterITCase.java b/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriterITCase.java
index c071721..84767e8 100644
--- a/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriterITCase.java
+++ b/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriterITCase.java
@@ -53,6 +53,7 @@ import java.util.Optional;
import static org.apache.flink.connector.mongodb.testutils.MongoTestUtil.assertThatIdsAreNotWritten;
import static org.apache.flink.connector.mongodb.testutils.MongoTestUtil.assertThatIdsAreWritten;
+import static org.apache.flink.connector.mongodb.testutils.MongoTestUtil.assertThatIdsAreWrittenWithMaxWaitTime;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
@@ -138,12 +139,12 @@ public class MongoWriterITCase {
createWriter(collection, batchSize, batchIntervalMs, flushOnCheckpoint)) {
writer.write(buildMessage(1), null);
writer.write(buildMessage(2), null);
+ writer.doBulkWrite();
writer.write(buildMessage(3), null);
writer.write(buildMessage(4), null);
- writer.doBulkWrite();
- }
- assertThatIdsAreWritten(collectionOf(collection), 1, 2, 3, 4);
+ assertThatIdsAreWrittenWithMaxWaitTime(collectionOf(collection), 10000L, 1, 2, 3, 4);
+ }
}
@Test
diff --git a/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/testutils/MongoTestUtil.java b/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/testutils/MongoTestUtil.java
index 7d8c3d0..246f2bc 100644
--- a/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/testutils/MongoTestUtil.java
+++ b/flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/testutils/MongoTestUtil.java
@@ -77,4 +77,17 @@ public class MongoTestUtil {
assertThat(actualIds).containsExactlyInAnyOrder(ids);
}
+
+ public static void assertThatIdsAreWrittenWithMaxWaitTime(
+ MongoCollection<Document> coll, long maxWaitTimeMs, Integer... ids)
+ throws InterruptedException {
+ long startTimeMillis = System.currentTimeMillis();
+ while (System.currentTimeMillis() - startTimeMillis < maxWaitTimeMs) {
+ if (coll.countDocuments(Filters.in("_id", ids)) == ids.length) {
+ break;
+ }
+ Thread.sleep(1000L);
+ }
+ assertThatIdsAreWritten(coll, ids);
+ }
}
[flink-connector-mongodb] 05/05: [FLINK-32348][connectors/mongodb][tests] Fix MongoDB tests are flaky and time out
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch v1.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-mongodb.git
commit 723b34a56012bf857a5e7da540adfe12a44bfe4c
Author: Jiabao Sun <ji...@xtransfer.cn>
AuthorDate: Wed Jul 12 17:12:06 2023 +0800
[FLINK-32348][connectors/mongodb][tests] Fix MongoDB tests are flaky and time out
This closes #13.
---
.../mongodb/source/enumerator/MongoSourceEnumerator.java | 13 +++++++++----
.../source/reader/split/MongoScanSourceSplitReader.java | 1 -
2 files changed, 9 insertions(+), 5 deletions(-)
diff --git a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/enumerator/MongoSourceEnumerator.java b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/enumerator/MongoSourceEnumerator.java
index d13a843..951c527 100644
--- a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/enumerator/MongoSourceEnumerator.java
+++ b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/enumerator/MongoSourceEnumerator.java
@@ -97,6 +97,15 @@ public class MongoSourceEnumerator
continue;
}
+ // close idle readers
+ if (splitAssigner.noMoreSplits() && boundedness == Boundedness.BOUNDED) {
+ context.signalNoMoreSplits(nextAwaiting);
+ awaitingReader.remove();
+ LOG.info(
+ "All scan splits have been assigned, closing idle reader {}", nextAwaiting);
+ continue;
+ }
+
Optional<MongoSourceSplit> split = splitAssigner.getNext();
if (split.isPresent()) {
final MongoSourceSplit mongoSplit = split.get();
@@ -104,10 +113,6 @@ public class MongoSourceEnumerator
awaitingReader.remove();
LOG.info("Assign split {} to subtask {}", mongoSplit, nextAwaiting);
break;
- } else if (splitAssigner.noMoreSplits() && boundedness == Boundedness.BOUNDED) {
- LOG.info("All splits have been assigned");
- context.registeredReaders().keySet().forEach(context::signalNoMoreSplits);
- break;
} else {
// there is no available splits by now, skip assigning
break;
diff --git a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/reader/split/MongoScanSourceSplitReader.java b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/reader/split/MongoScanSourceSplitReader.java
index 4702f94..134fe73 100644
--- a/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/reader/split/MongoScanSourceSplitReader.java
+++ b/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/reader/split/MongoScanSourceSplitReader.java
@@ -119,7 +119,6 @@ public class MongoScanSourceSplitReader implements MongoSourceSplitReader<MongoS
throw new IOException("Scan records form MongoDB failed", e);
} finally {
if (finished) {
- currentSplit = null;
closeCursor();
}
}
[flink-connector-mongodb] 02/05: [FLINK-31923] Run nightly builds against multiple branches and Flink versions
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch v1.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-mongodb.git
commit 99755b44f9964cecd066641d2147aa134ed41b00
Author: Martijn Visser <mv...@confluent.io>
AuthorDate: Thu Jun 15 10:54:10 2023 +0200
[FLINK-31923] Run nightly builds against multiple branches and Flink versions
---
.github/workflows/weekly.yml | 24 ++++++++++++++++++++++--
1 file changed, 22 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/weekly.yml b/.github/workflows/weekly.yml
index 577ff1f..bba17f1 100644
--- a/.github/workflows/weekly.yml
+++ b/.github/workflows/weekly.yml
@@ -26,7 +26,27 @@ jobs:
if: github.repository_owner == 'apache'
strategy:
matrix:
- flink: [1.16-SNAPSHOT, 1.17-SNAPSHOT]
+ flink_branches: [{
+ flink: 1.16-SNAPSHOT,
+ branch: main
+ }, {
+ flink: 1.17-SNAPSHOT,
+ branch: main
+ }, {
+ flink: 1.18-SNAPSHOT,
+ branch: main
+ }, {
+ flink: 1.16.2,
+ branch: v1.0
+ }, {
+ flink: 1.17.1,
+ branch: v1.0
+ }, {
+ flink: 1.18-SNAPSHOT,
+ branch: v1.0
+ }]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
- flink_version: ${{ matrix.flink }}
+ flink_version: ${{ matrix.flink_branches.flink }}
+ connector_branch: ${{ matrix.flink_branches.branch }}
+ run_dependency_convergence: false