Merge branch 'master' into regex

Author: whhe

docs/content/connectors/mongodb-cdc.md

@@ -226,6 +226,13 @@ Connector Options
       <td>Integer</td>
       <td>The max size of the queue to use when copying data.</td>
     </tr>
+    <tr>
+      <td>batch.size</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">0</td>
+      <td>Integer</td>
+      <td>Change stream cursor batch size. Specifies the maximum number of change events to return in each batch of the response from the MongoDB cluster. The default is 0 meaning it uses the server's default value.</td>
+    </tr>
     <tr>
       <td>poll.max.batch.size</td>
       <td>optional</td>

docs/content/connectors/sqlserver-cdc.md

@@ -26,6 +26,40 @@ Download [flink-sql-connector-sqlserver-cdc-2.3-SNAPSHOT.jar](https://repo1.mave
 
 **Note:** flink-sql-connector-sqlserver-cdc-XXX-SNAPSHOT version is the code corresponding to the development branch. Users need to download the source code and compile the corresponding jar. Users should use the released version, such as [flink-sql-connector-sqlserver-cdc-XXX.jar](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-sqlserver-cdc), the released version will be available in the Maven central warehouse.
 
+Setup SQLServer Database
+----------------
+A SQL Server administrator must enable change data capture on the source tables that you want to capture. The database must already be enabled for CDC. To enable CDC on a table, a SQL Server administrator runs the stored procedure ```sys.sp_cdc_enable_table``` for the table.
+
+**Prerequisites:**
+* CDC is enabled on the SQL Server database.
+* The SQL Server Agent is running.
+* You are a member of the db_owner fixed database role for the database.
+
+**Procedure:**
+* Connect to the SQL Server database by database management studio.
+* Run the following SQL statement to enable CDC on the table.
+```sql
+USE MyDB
+GO
+
+EXEC sys.sp_cdc_enable_table
+@source_schema = N'dbo',     -- Specifies the schema of the source table.
+@source_name   = N'MyTable', -- Specifies the name of the table that you want to capture.
+@role_name     = N'MyRole',  -- Specifies a role MyRole to which you can add users to whom you want to grant SELECT permission on the captured columns of the source table. Users in the sysadmin or db_owner role also have access to the specified change tables. Set the value of @role_name to NULL, to allow only members in the sysadmin or db_owner to have full access to captured information.
+@filegroup_name = N'MyDB_CT',-- Specifies the filegroup where SQL Server places the change table for the captured table. The named filegroup must already exist. It is best not to locate change tables in the same filegroup that you use for source tables.
+@supports_net_changes = 0
+GO
+```
+* Verifying that the user has access to the CDC table
+```sql
+--The following example runs the stored procedure sys.sp_cdc_help_change_data_capture on the database MyDB:
+USE MyDB;
+GO
+EXEC sys.sp_cdc_help_change_data_capture
+GO
+```
+The query returns configuration information for each table in the database that is enabled for CDC and that contains change data that the caller is authorized to access. If the result is empty, verify that the user has privileges to access both the capture instance and the CDC tables.
+
 How to create a SQLServer CDC table
 ----------------
 

flink-connector-mongodb-cdc/src/main/java/com/ververica/cdc/connectors/mongodb/MongoDBSource.java

@@ -59,6 +59,8 @@ public class MongoDBSource {
 
     public static final String FULL_DOCUMENT_UPDATE_LOOKUP = FullDocument.UPDATE_LOOKUP.getValue();
 
+    public static final int BATCH_SIZE_DEFAULT = 0;
+
     public static final int POLL_MAX_BATCH_SIZE_DEFAULT = 1000;
 
     public static final int POLL_AWAIT_TIME_MILLIS_DEFAULT = 1500;
@@ -128,7 +130,7 @@ public static class Builder<T> {
         private List<String> databaseList;
         private List<String> collectionList;
         private String connectionOptions;
-        private Integer batchSize;
+        private Integer batchSize = BATCH_SIZE_DEFAULT;
         private Integer pollAwaitTimeMillis = POLL_AWAIT_TIME_MILLIS_DEFAULT;
         private Integer pollMaxBatchSize = POLL_MAX_BATCH_SIZE_DEFAULT;
         private Boolean copyExisting = true;
@@ -186,7 +188,9 @@ public Builder<T> collectionList(String... collectionList) {
         /**
          * batch.size
          *
-         * <p>The cursor batch size. Default: 0
+         * <p>The change stream cursor batch size. Specifies the maximum number of change events to
+         * return in each batch of the response from the MongoDB cluster. The default is 0 meaning
+         * it uses the server's default value. Default: 0
          */
         public Builder<T> batchSize(int batchSize) {
             checkArgument(batchSize >= 0);

flink-connector-mongodb-cdc/src/main/java/com/ververica/cdc/connectors/mongodb/table/MongoDBTableSource.java

@@ -69,6 +69,7 @@ public class MongoDBTableSource implements ScanTableSource, SupportsReadingMetad
     private final String copyExistingPipeline;
     private final Integer copyExistingMaxThreads;
     private final Integer copyExistingQueueSize;
+    private final Integer batchSize;
     private final Integer pollMaxBatchSize;
     private final Integer pollAwaitTimeMillis;
     private final Integer heartbeatIntervalMillis;
@@ -98,6 +99,7 @@ public MongoDBTableSource(
             @Nullable String copyExistingPipeline,
             @Nullable Integer copyExistingMaxThreads,
             @Nullable Integer copyExistingQueueSize,
+            @Nullable Integer batchSize,
             @Nullable Integer pollMaxBatchSize,
             @Nullable Integer pollAwaitTimeMillis,
             @Nullable Integer heartbeatIntervalMillis,
@@ -115,6 +117,7 @@ public MongoDBTableSource(
         this.copyExistingPipeline = copyExistingPipeline;
         this.copyExistingMaxThreads = copyExistingMaxThreads;
         this.copyExistingQueueSize = copyExistingQueueSize;
+        this.batchSize = batchSize;
         this.pollMaxBatchSize = pollMaxBatchSize;
         this.pollAwaitTimeMillis = pollAwaitTimeMillis;
         this.heartbeatIntervalMillis = heartbeatIntervalMillis;
@@ -175,6 +178,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
         Optional.ofNullable(copyExistingPipeline).ifPresent(builder::copyExistingPipeline);
         Optional.ofNullable(copyExistingMaxThreads).ifPresent(builder::copyExistingMaxThreads);
         Optional.ofNullable(copyExistingQueueSize).ifPresent(builder::copyExistingQueueSize);
+        Optional.ofNullable(batchSize).ifPresent(builder::batchSize);
         Optional.ofNullable(pollMaxBatchSize).ifPresent(builder::pollMaxBatchSize);
         Optional.ofNullable(pollAwaitTimeMillis).ifPresent(builder::pollAwaitTimeMillis);
         Optional.ofNullable(heartbeatIntervalMillis).ifPresent(builder::heartbeatIntervalMillis);
@@ -232,6 +236,7 @@ public DynamicTableSource copy() {
                         copyExistingPipeline,
                         copyExistingMaxThreads,
                         copyExistingQueueSize,
+                        batchSize,
                         pollMaxBatchSize,
                         pollAwaitTimeMillis,
                         heartbeatIntervalMillis,
@@ -263,6 +268,7 @@ public boolean equals(Object o) {
                 && Objects.equals(copyExistingPipeline, that.copyExistingPipeline)
                 && Objects.equals(copyExistingMaxThreads, that.copyExistingMaxThreads)
                 && Objects.equals(copyExistingQueueSize, that.copyExistingQueueSize)
+                && Objects.equals(batchSize, that.batchSize)
                 && Objects.equals(pollMaxBatchSize, that.pollMaxBatchSize)
                 && Objects.equals(pollAwaitTimeMillis, that.pollAwaitTimeMillis)
                 && Objects.equals(heartbeatIntervalMillis, that.heartbeatIntervalMillis)
@@ -287,6 +293,7 @@ public int hashCode() {
                 copyExistingPipeline,
                 copyExistingMaxThreads,
                 copyExistingQueueSize,
+                batchSize,
                 pollMaxBatchSize,
                 pollAwaitTimeMillis,
                 heartbeatIntervalMillis,

flink-connector-mongodb-cdc/src/main/java/com/ververica/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java

@@ -30,6 +30,7 @@
 import java.util.HashSet;
 import java.util.Set;
 
+import static com.ververica.cdc.connectors.mongodb.MongoDBSource.BATCH_SIZE_DEFAULT;
 import static com.ververica.cdc.connectors.mongodb.MongoDBSource.ERROR_TOLERANCE_NONE;
 import static com.ververica.cdc.connectors.mongodb.MongoDBSource.POLL_AWAIT_TIME_MILLIS_DEFAULT;
 import static com.ververica.cdc.connectors.mongodb.MongoDBSource.POLL_MAX_BATCH_SIZE_DEFAULT;
@@ -148,6 +149,16 @@ public class MongoDBTableSourceFactory implements DynamicTableSourceFactory {
                     .withDescription(
                             "The max size of the queue to use when copying data. Defaults to 16000.");
 
+    private static final ConfigOption<Integer> BATCH_SIZE =
+            ConfigOptions.key("batch.size")
+                    .intType()
+                    .defaultValue(BATCH_SIZE_DEFAULT)
+                    .withDescription(
+                            "Change stream cursor batch size. "
+                                    + "Specifies the maximum number of change events to return in each batch "
+                                    + "of the response from the MongoDB cluster."
+                                    + "Defaults to 0 meaning it uses the server's default value.");
+
     private static final ConfigOption<Integer> POLL_MAX_BATCH_SIZE =
             ConfigOptions.key("poll.max.batch.size")
                     .intType()
@@ -196,6 +207,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
         String errorsTolerance = config.get(ERRORS_TOLERANCE);
         Boolean errorsLogEnable = config.get(ERRORS_LOG_ENABLE);
 
+        Integer batchSize = config.get(BATCH_SIZE);
         Integer pollMaxBatchSize = config.get(POLL_MAX_BATCH_SIZE);
         Integer pollAwaitTimeMillis = config.get(POLL_AWAIT_TIME_MILLIS);
 
@@ -232,6 +244,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 copyExistingPipeline,
                 copyExistingMaxThreads,
                 copyExistingQueueSize,
+                batchSize,
                 pollMaxBatchSize,
                 pollAwaitTimeMillis,
                 heartbeatIntervalMillis,
@@ -270,6 +283,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(COPY_EXISTING_PIPELINE);
         options.add(COPY_EXISTING_MAX_THREADS);
         options.add(COPY_EXISTING_QUEUE_SIZE);
+        options.add(BATCH_SIZE);
         options.add(POLL_MAX_BATCH_SIZE);
         options.add(POLL_AWAIT_TIME_MILLIS);
         options.add(HEARTBEAT_INTERVAL_MILLIS);

flink-connector-mongodb-cdc/src/test/java/com/ververica/cdc/connectors/mongodb/MongoDBSourceExampleTest.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2022 Ververica Inc.
+ *
+ * Licensed 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 com.ververica.cdc.connectors.mongodb;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
+import org.junit.Ignore;
+import org.junit.Test;
+
+/** Example Tests for {@link MongoDBSource}. */
+public class MongoDBSourceExampleTest extends MongoDBTestBase {
+
+    @Test
+    @Ignore("Test ignored because it won't stop and is used for manual test")
+    public void testConsumingAllEvents() throws Exception {
+        String inventory = executeCommandFileInSeparateDatabase("inventory");
+
+        SourceFunction<String> sourceFunction =
+                MongoDBSource.<String>builder()
+                        .hosts(MONGODB_CONTAINER.getHostAndPort())
+                        .username(FLINK_USER)
+                        .password(FLINK_USER_PASSWORD)
+                        .databaseList(inventory)
+                        .collectionList(inventory + ".products")
+                        .deserializer(new JsonDebeziumDeserializationSchema())
+                        .build();
+
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        env.addSource(sourceFunction)
+                .print()
+                .setParallelism(1); // use parallelism 1 for sink to keep message ordering
+
+        env.execute("Print MongoDB Snapshot + Change Stream Events");
+    }
+}

flink-connector-mongodb-cdc/src/test/java/com/ververica/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java

@@ -44,6 +44,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import static com.ververica.cdc.connectors.mongodb.MongoDBSource.BATCH_SIZE_DEFAULT;
 import static com.ververica.cdc.connectors.mongodb.MongoDBSource.ERROR_TOLERANCE_ALL;
 import static com.ververica.cdc.connectors.mongodb.MongoDBSource.POLL_AWAIT_TIME_MILLIS_DEFAULT;
 import static com.ververica.cdc.connectors.mongodb.MongoDBSource.POLL_MAX_BATCH_SIZE_DEFAULT;
@@ -110,6 +111,7 @@ public void testCommonProperties() {
                         null,
                         null,
                         null,
+                        BATCH_SIZE_DEFAULT,
                         POLL_MAX_BATCH_SIZE_DEFAULT,
                         POLL_AWAIT_TIME_MILLIS_DEFAULT,
                         null,
@@ -127,6 +129,7 @@ public void testOptionalProperties() {
         options.put("copy.existing.pipeline", "[ { \"$match\": { \"closed\": \"false\" } } ]");
         options.put("copy.existing.max.threads", "1");
         options.put("copy.existing.queue.size", "101");
+        options.put("batch.size", "101");
         options.put("poll.max.batch.size", "102");
         options.put("poll.await.time.ms", "103");
         options.put("heartbeat.interval.ms", "104");
@@ -147,6 +150,7 @@ public void testOptionalProperties() {
                         "[ { \"$match\": { \"closed\": \"false\" } } ]",
                         1,
                         101,
+                        101,
                         102,
                         103,
                         104,
@@ -181,6 +185,7 @@ public void testMetadataColumns() {
                         null,
                         null,
                         null,
+                        BATCH_SIZE_DEFAULT,
                         POLL_MAX_BATCH_SIZE_DEFAULT,
                         POLL_AWAIT_TIME_MILLIS_DEFAULT,
                         null,