You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/06/10 10:16:30 UTC
[GitHub] [pulsar] shink opened a new pull request, #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
shink opened a new pull request, #16003:
URL: https://github.com/apache/pulsar/pull/16003
### Motivation
Now, the MongoDB source connector only supports the incremental message synchronization.
This PR adds support for full message synchronization.
Since MongDB 4.0, we can set the starting point for the change stream by the `startAtOperationTime` field.
So, we can set it to `0` to make start point the earliest.
See https://www.mongodb.com/docs/v4.2/reference/method/db.collection.watch/ for more information.
### Modifications
1. Improve config object.
There are some commonalities and differences between sink configuration and source configuration.
So, I created an abstract class called `MongoAbstractConfig` which contains the commonalities between them.
MongoSourceConfig` and `MongoSinkConfig` contain the unique content.
2. Add support for full message synchronization in the source connector.
```java
if (SyncType.FULL_SYNC.equals(mongoSourceConfig.getSyncType())) {
// sync currently existing messages
// startAtOperationTime is the starting point for the change stream
// setting startAtOperationTime to 0 means the start point is the earliest
// see https://www.mongodb.com/docs/v4.2/reference/method/db.collection.watch/ for more information
stream.startAtOperationTime(new BsonTimestamp(0L));
}
```
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
No.
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [x] `doc-required`
(Your PR needs to update docs and you will update later)
- [ ] `doc-not-needed`
(Please explain why)
- [ ] `doc`
(Your PR contains doc changes)
- [ ] `doc-complete`
(Docs have been already added)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1204677247
The pr had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1152212638
I will finish UT later. Convert this PR to a draft temporarily.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
eolivelli commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1221580180
/pulsar-bot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
codelipenghui merged PR #16003:
URL: https://github.com/apache/pulsar/pull/16003
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1153128898
@315157973 PTAL 🎉
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1221541856
@codelipenghui Hi, it seems that all the failed checks are flask tests, what shhould I do?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1161408072
@315157973 OK. And where should I add the doc to?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1210797903
/pulsarbot run-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
poorbarcode commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1249987498
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1250204320
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1250195778
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on a diff in pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on code in PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#discussion_r894377745
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSource.java:
##########
@@ -79,38 +78,46 @@ public MongoSource(Supplier<MongoClient> clientProvider) {
public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
log.info("Open MongoDB Source");
- mongoConfig = MongoConfig.load(config);
- mongoConfig.validate(false, false);
+ mongoSourceConfig = MongoSourceConfig.load(config);
+ mongoSourceConfig.validate();
if (clientProvider != null) {
mongoClient = clientProvider.get();
} else {
- mongoClient = MongoClients.create(mongoConfig.getMongoUri());
+ mongoClient = MongoClients.create(mongoSourceConfig.getMongoUri());
}
- if (StringUtils.isEmpty(mongoConfig.getDatabase())) {
+ if (StringUtils.isEmpty(mongoSourceConfig.getDatabase())) {
// Watch all databases
log.info("Watch all");
stream = mongoClient.watch();
} else {
- final MongoDatabase db = mongoClient.getDatabase(mongoConfig.getDatabase());
+ final MongoDatabase db = mongoClient.getDatabase(mongoSourceConfig.getDatabase());
- if (StringUtils.isEmpty(mongoConfig.getCollection())) {
+ if (StringUtils.isEmpty(mongoSourceConfig.getCollection())) {
// Watch all collections in a database
log.info("Watch db: {}", db.getName());
stream = db.watch();
} else {
// Watch a collection
-
- final MongoCollection<Document> collection = db.getCollection(mongoConfig.getCollection());
- log.info("Watch collection: {} {}", db.getName(), mongoConfig.getCollection());
+ final MongoCollection<Document> collection = db.getCollection(mongoSourceConfig.getCollection());
+ log.info("Watch collection: {}.{}", db.getName(), mongoSourceConfig.getCollection());
stream = collection.watch();
}
}
- stream.batchSize(mongoConfig.getBatchSize()).fullDocument(FullDocument.UPDATE_LOOKUP);
+ stream.batchSize(mongoSourceConfig.getBatchSize())
+ .fullDocument(FullDocument.UPDATE_LOOKUP);
+
+ if (SyncType.FULL_SYNC.equals(mongoSourceConfig.getSyncType())) {
+ // sync currently existing messages
+ // startAtOperationTime is the starting point for the change stream
+ // setting startAtOperationTime to 0 means the start point is the earliest
+ // see https://www.mongodb.com/docs/v4.2/reference/method/db.collection.watch/ for more information
+ stream.startAtOperationTime(new BsonTimestamp(0L));
+ }
Review Comment:
Here is the key.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] 315157973 commented on a diff in pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
315157973 commented on code in PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#discussion_r902206842
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSource.java:
##########
@@ -79,38 +78,46 @@ public MongoSource(Supplier<MongoClient> clientProvider) {
public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
log.info("Open MongoDB Source");
- mongoConfig = MongoConfig.load(config);
- mongoConfig.validate(false, false);
+ mongoSourceConfig = MongoSourceConfig.load(config);
+ mongoSourceConfig.validate();
if (clientProvider != null) {
mongoClient = clientProvider.get();
} else {
- mongoClient = MongoClients.create(mongoConfig.getMongoUri());
+ mongoClient = MongoClients.create(mongoSourceConfig.getMongoUri());
}
- if (StringUtils.isEmpty(mongoConfig.getDatabase())) {
+ if (StringUtils.isEmpty(mongoSourceConfig.getDatabase())) {
// Watch all databases
log.info("Watch all");
stream = mongoClient.watch();
} else {
- final MongoDatabase db = mongoClient.getDatabase(mongoConfig.getDatabase());
+ final MongoDatabase db = mongoClient.getDatabase(mongoSourceConfig.getDatabase());
- if (StringUtils.isEmpty(mongoConfig.getCollection())) {
+ if (StringUtils.isEmpty(mongoSourceConfig.getCollection())) {
// Watch all collections in a database
log.info("Watch db: {}", db.getName());
stream = db.watch();
} else {
// Watch a collection
-
- final MongoCollection<Document> collection = db.getCollection(mongoConfig.getCollection());
- log.info("Watch collection: {} {}", db.getName(), mongoConfig.getCollection());
+ final MongoCollection<Document> collection = db.getCollection(mongoSourceConfig.getCollection());
+ log.info("Watch collection: {}.{}", db.getName(), mongoSourceConfig.getCollection());
stream = collection.watch();
}
}
- stream.batchSize(mongoConfig.getBatchSize()).fullDocument(FullDocument.UPDATE_LOOKUP);
+ stream.batchSize(mongoSourceConfig.getBatchSize())
+ .fullDocument(FullDocument.UPDATE_LOOKUP);
+
+ if (SyncType.FULL_SYNC.equals(mongoSourceConfig.getSyncType())) {
Review Comment:
use `==` instead of `equals`
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSourceConfig.java:
##########
@@ -0,0 +1,102 @@
+/**
+ * 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.pulsar.io.mongodb;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+/**
+ * Configuration class for the MongoDB Source Connectors.
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class MongoSourceConfig extends MongoAbstractConfig {
+
+ private static final long serialVersionUID = 1152890092264945317L;
+
+ public static final SyncType DEFAULT_SYNC_TYPE = SyncType.INCR_SYNC;
+
+ public static final String DEFAULT_SYNC_TYPE_STR = "incr";
+
+ @FieldDoc(
+ defaultValue = DEFAULT_SYNC_TYPE_STR,
+ help = "The message synchronization type of the source connector. "
+ + "The field values can be of two types: incr and full. "
+ + "When it is set to incr, the source connector will only watch for changes made from now on. "
+ + "When it is set to full, the source connector will synchronize currently existing messages "
+ + "and watch for future changes."
+ )
+ private SyncType syncType = DEFAULT_SYNC_TYPE;
+
+ public static MongoSourceConfig load(String yamlFile) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ final MongoSourceConfig cfg = mapper.readValue(new File(yamlFile), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ public static MongoSourceConfig load(Map<String, Object> map) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper();
+ final MongoSourceConfig cfg =
+ mapper.readValue(new ObjectMapper().writeValueAsString(map), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ /**
+ * Override syncType setter method.
+ *
+ * @param syncType Sync type string.
+ */
+ public void setSyncType(String syncType) {
Review Comment:
Why not use `@JsonCreator` ?
The type string must be consistent with the enumeration. If the type string cannot be parsed, an exception must be thrown.
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSourceConfig.java:
##########
@@ -0,0 +1,102 @@
+/**
+ * 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.pulsar.io.mongodb;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+/**
+ * Configuration class for the MongoDB Source Connectors.
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class MongoSourceConfig extends MongoAbstractConfig {
+
+ private static final long serialVersionUID = 1152890092264945317L;
+
+ public static final SyncType DEFAULT_SYNC_TYPE = SyncType.INCR_SYNC;
+
+ public static final String DEFAULT_SYNC_TYPE_STR = "incr";
+
+ @FieldDoc(
+ defaultValue = DEFAULT_SYNC_TYPE_STR,
+ help = "The message synchronization type of the source connector. "
+ + "The field values can be of two types: incr and full. "
+ + "When it is set to incr, the source connector will only watch for changes made from now on. "
+ + "When it is set to full, the source connector will synchronize currently existing messages "
+ + "and watch for future changes."
+ )
+ private SyncType syncType = DEFAULT_SYNC_TYPE;
+
+ public static MongoSourceConfig load(String yamlFile) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ final MongoSourceConfig cfg = mapper.readValue(new File(yamlFile), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ public static MongoSourceConfig load(Map<String, Object> map) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper();
+ final MongoSourceConfig cfg =
+ mapper.readValue(new ObjectMapper().writeValueAsString(map), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ /**
+ * Override syncType setter method.
+ *
+ * @param syncType Sync type string.
+ */
+ public void setSyncType(String syncType) {
+ if ("full".equalsIgnoreCase(syncType)
+ || "full_sync".equalsIgnoreCase(syncType)
+ || "full-sync".equalsIgnoreCase(syncType)) {
+
+ this.syncType = SyncType.FULL_SYNC;
+ return;
+ }
+
+ if ("incr".equalsIgnoreCase(syncType)
+ || "increment".equalsIgnoreCase(syncType)
+ || "incr_sync".equalsIgnoreCase(syncType)
+ || "incr-sync".equalsIgnoreCase(syncType)) {
+
+ this.syncType = SyncType.INCR_SYNC;
+ return;
+ }
+
+ this.syncType = DEFAULT_SYNC_TYPE;
+ }
+
+ @Override
+ public void validate() {
+ super.validate();
+ checkNotNull(getSyncType(), "syncType not set.");
Review Comment:
This should never be null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] 315157973 commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
315157973 commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1161346117
Please add some doc and make sure that the change passes the CI checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1210872125
@codelipenghui Thank you very much! But why are there three failed checks after a rerun, and before there were four. And I didn't make changes to broker.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on a diff in pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on code in PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#discussion_r894377258
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSource.java:
##########
@@ -51,19 +52,17 @@
name = "mongo",
type = IOType.SOURCE,
help = "A source connector that sends mongodb documents to pulsar",
- configClass = MongoConfig.class
+ configClass = MongoSourceConfig.class
)
@Slf4j
public class MongoSource extends PushSource<byte[]> {
private final Supplier<MongoClient> clientProvider;
- private MongoConfig mongoConfig;
+ private MongoSourceConfig mongoSourceConfig;
private MongoClient mongoClient;
- private Thread streamThread;
-
Review Comment:
It is never used. So I delete it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
poorbarcode commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1249999983
Hi @shink
> https://github.com/apache/pulsar/actions/runs/3071925540/jobs/4963093961
this is a flaky test, It should be fine if it is run several times
> https://github.com/apache/pulsar/actions/runs/3071925517/jobs/4963032830
I read the log and found that instability was caused by `mockito`. It should be fine if it is run several times
Many tests in this group(`Pulsar CI Flaky`) are unstable, but this does not affect merge. These problems are being fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1250000247
@poorbarcode Thank you so much for your help!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1249998892
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] momo-jun commented on pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
momo-jun commented on PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#issuecomment-1255870427
@shink seems you've already included docs in this PR, so I will update the `doc-required` label to `doc`. Thank you for adding the docs!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on a diff in pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on code in PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#discussion_r894378935
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSourceConfig.java:
##########
@@ -0,0 +1,102 @@
+/**
+ * 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.pulsar.io.mongodb;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.google.common.base.Preconditions;
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+/**
+ * Configuration class for the MongoDB Source Connectors.
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class MongoSourceConfig extends MongoAbstractConfig {
+
+ private static final long serialVersionUID = 1152890092264945317L;
+
+ public static final SyncType DEFAULT_SYNC_TYPE = SyncType.INCR_SYNC;
+
+ public static final String DEFAULT_SYNC_TYPE_STR = "incr";
+
+ @FieldDoc(
+ defaultValue = DEFAULT_SYNC_TYPE_STR,
+ help = "The message synchronization type of the source connector. "
+ + "The field values can be of two types: incr and full. "
+ + "When it is set to incr, the source connector will only watch for changes made from now on. "
+ + "When it is set to full, the source connector will synchronize currently existing messages "
+ + "and watch for future changes."
+ )
+ private SyncType syncType = DEFAULT_SYNC_TYPE;
+
+ public static MongoSourceConfig load(String yamlFile) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ final MongoSourceConfig cfg = mapper.readValue(new File(yamlFile), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ public static MongoSourceConfig load(Map<String, Object> map) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper();
+ final MongoSourceConfig cfg =
+ mapper.readValue(new ObjectMapper().writeValueAsString(map), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ /**
+ * Override syncType setter method.
+ *
+ * @param syncType Sync type string.
+ */
+ public void setSyncType(String syncType) {
+ if ("full".equalsIgnoreCase(syncType)
+ || "full_sync".equalsIgnoreCase(syncType)
+ || "full-sync".equalsIgnoreCase(syncType)) {
+
+ this.syncType = SyncType.FULL_SYNC;
+ return;
+ }
+
+ if ("incr".equalsIgnoreCase(syncType)
+ || "increment".equalsIgnoreCase(syncType)
+ || "incr_sync".equalsIgnoreCase(syncType)
+ || "incr-sync".equalsIgnoreCase(syncType)) {
+
+ this.syncType = SyncType.INCR_SYNC;
+ return;
+ }
+
+ this.syncType = DEFAULT_SYNC_TYPE;
+ }
Review Comment:
Is the deserialization method of `syncType` OK?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on a diff in pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on code in PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#discussion_r902285585
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSourceConfig.java:
##########
@@ -0,0 +1,102 @@
+/**
+ * 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.pulsar.io.mongodb;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+/**
+ * Configuration class for the MongoDB Source Connectors.
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class MongoSourceConfig extends MongoAbstractConfig {
+
+ private static final long serialVersionUID = 1152890092264945317L;
+
+ public static final SyncType DEFAULT_SYNC_TYPE = SyncType.INCR_SYNC;
+
+ public static final String DEFAULT_SYNC_TYPE_STR = "incr";
+
+ @FieldDoc(
+ defaultValue = DEFAULT_SYNC_TYPE_STR,
+ help = "The message synchronization type of the source connector. "
+ + "The field values can be of two types: incr and full. "
+ + "When it is set to incr, the source connector will only watch for changes made from now on. "
+ + "When it is set to full, the source connector will synchronize currently existing messages "
+ + "and watch for future changes."
+ )
+ private SyncType syncType = DEFAULT_SYNC_TYPE;
+
+ public static MongoSourceConfig load(String yamlFile) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ final MongoSourceConfig cfg = mapper.readValue(new File(yamlFile), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ public static MongoSourceConfig load(Map<String, Object> map) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper();
+ final MongoSourceConfig cfg =
+ mapper.readValue(new ObjectMapper().writeValueAsString(map), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ /**
+ * Override syncType setter method.
+ *
+ * @param syncType Sync type string.
+ */
+ public void setSyncType(String syncType) {
+ if ("full".equalsIgnoreCase(syncType)
+ || "full_sync".equalsIgnoreCase(syncType)
+ || "full-sync".equalsIgnoreCase(syncType)) {
+
+ this.syncType = SyncType.FULL_SYNC;
+ return;
+ }
+
+ if ("incr".equalsIgnoreCase(syncType)
+ || "increment".equalsIgnoreCase(syncType)
+ || "incr_sync".equalsIgnoreCase(syncType)
+ || "incr-sync".equalsIgnoreCase(syncType)) {
+
+ this.syncType = SyncType.INCR_SYNC;
+ return;
+ }
+
+ this.syncType = DEFAULT_SYNC_TYPE;
+ }
+
+ @Override
+ public void validate() {
+ super.validate();
+ checkNotNull(getSyncType(), "syncType not set.");
Review Comment:
@315157973 Yes, it is. This is redundant. I will remove it in the next commit.
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSource.java:
##########
@@ -79,38 +78,46 @@ public MongoSource(Supplier<MongoClient> clientProvider) {
public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
log.info("Open MongoDB Source");
- mongoConfig = MongoConfig.load(config);
- mongoConfig.validate(false, false);
+ mongoSourceConfig = MongoSourceConfig.load(config);
+ mongoSourceConfig.validate();
if (clientProvider != null) {
mongoClient = clientProvider.get();
} else {
- mongoClient = MongoClients.create(mongoConfig.getMongoUri());
+ mongoClient = MongoClients.create(mongoSourceConfig.getMongoUri());
}
- if (StringUtils.isEmpty(mongoConfig.getDatabase())) {
+ if (StringUtils.isEmpty(mongoSourceConfig.getDatabase())) {
// Watch all databases
log.info("Watch all");
stream = mongoClient.watch();
} else {
- final MongoDatabase db = mongoClient.getDatabase(mongoConfig.getDatabase());
+ final MongoDatabase db = mongoClient.getDatabase(mongoSourceConfig.getDatabase());
- if (StringUtils.isEmpty(mongoConfig.getCollection())) {
+ if (StringUtils.isEmpty(mongoSourceConfig.getCollection())) {
// Watch all collections in a database
log.info("Watch db: {}", db.getName());
stream = db.watch();
} else {
// Watch a collection
-
- final MongoCollection<Document> collection = db.getCollection(mongoConfig.getCollection());
- log.info("Watch collection: {} {}", db.getName(), mongoConfig.getCollection());
+ final MongoCollection<Document> collection = db.getCollection(mongoSourceConfig.getCollection());
+ log.info("Watch collection: {}.{}", db.getName(), mongoSourceConfig.getCollection());
stream = collection.watch();
}
}
- stream.batchSize(mongoConfig.getBatchSize()).fullDocument(FullDocument.UPDATE_LOOKUP);
+ stream.batchSize(mongoSourceConfig.getBatchSize())
+ .fullDocument(FullDocument.UPDATE_LOOKUP);
+
+ if (SyncType.FULL_SYNC.equals(mongoSourceConfig.getSyncType())) {
Review Comment:
@315157973 Yes, using `==` is indeed better. I will do it in the next commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shink commented on a diff in pull request #16003: [feature][pulsar-io-mongo] Add support for full message synchronization
Posted by GitBox <gi...@apache.org>.
shink commented on code in PR #16003:
URL: https://github.com/apache/pulsar/pull/16003#discussion_r902284001
##########
pulsar-io/mongo/src/main/java/org/apache/pulsar/io/mongodb/MongoSourceConfig.java:
##########
@@ -0,0 +1,102 @@
+/**
+ * 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.pulsar.io.mongodb;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+/**
+ * Configuration class for the MongoDB Source Connectors.
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class MongoSourceConfig extends MongoAbstractConfig {
+
+ private static final long serialVersionUID = 1152890092264945317L;
+
+ public static final SyncType DEFAULT_SYNC_TYPE = SyncType.INCR_SYNC;
+
+ public static final String DEFAULT_SYNC_TYPE_STR = "incr";
+
+ @FieldDoc(
+ defaultValue = DEFAULT_SYNC_TYPE_STR,
+ help = "The message synchronization type of the source connector. "
+ + "The field values can be of two types: incr and full. "
+ + "When it is set to incr, the source connector will only watch for changes made from now on. "
+ + "When it is set to full, the source connector will synchronize currently existing messages "
+ + "and watch for future changes."
+ )
+ private SyncType syncType = DEFAULT_SYNC_TYPE;
+
+ public static MongoSourceConfig load(String yamlFile) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ final MongoSourceConfig cfg = mapper.readValue(new File(yamlFile), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ public static MongoSourceConfig load(Map<String, Object> map) throws IOException {
+ final ObjectMapper mapper = new ObjectMapper();
+ final MongoSourceConfig cfg =
+ mapper.readValue(new ObjectMapper().writeValueAsString(map), MongoSourceConfig.class);
+
+ return cfg;
+ }
+
+ /**
+ * Override syncType setter method.
+ *
+ * @param syncType Sync type string.
+ */
+ public void setSyncType(String syncType) {
Review Comment:
@315157973
Functionally, I think that using a no-argument constructor and a `setter` method as same as using `@JsonCreater`. So I overrided the `setter` method of `syncType`.
Using `@JsonCreator` can express more clearly. But it will cause `MongoAbstractConfig`, `MongoSourceConfig`, and `MongoSinkConfig` to add constructors with arguments (Because the `@Data` annotation only adds a no-argument constractor).
BTW, is this deserialization method OK?
```java
this.syncType = SyncType.valueOf(syncTypeStr.toUpperCase());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org