Merge branch 'master' into universal-chunk-column-dev

Author: xxntti3n

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

@@ -271,18 +271,10 @@ private void captureNewlyAddedTables() {
 
                 // case 2: there are new tables to add
                 if (!newlyAddedTables.isEmpty()) {
-                    // if job is still in snapshot reading phase, directly add all newly added
-                    // tables
                     LOG.info("Found newly added tables, start capture newly added tables process");
 
-                    // add new tables
                     remainingTables.addAll(newlyAddedTables);
-                    if (AssignerStatus.isAssigningFinished(assignerStatus)) {
-                        // start the newly added tables process under binlog reading phase
-                        LOG.info(
-                                "Found newly added tables, start capture newly added tables process under binlog reading phase");
-                        this.startAssignNewlyAddedTables();
-                    }
+                    this.startAssignNewlyAddedTables();
                 }
             } catch (Exception e) {
                 throw new FlinkRuntimeException(

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java

@@ -135,6 +135,15 @@ public Map<ObjectPath, String> getChunkKeyColumns() {
     public boolean includePartitionedTables() {
         return includePartitionedTables;
     }
+    /**
+     * Returns {@code includePartitionedTables} value.
+     *
+     * @return include partitioned table
+     */
+    public boolean includePartitionedTables() {
+        return includePartitionedTables;
+    }
+
     /**
      * Returns the slot name for backfill task.
      *

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java

@@ -57,6 +57,7 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory {
 
     private Map<ObjectPath, String> chunkKeyColumns = new HashMap<>();
 
+
     private boolean includePartitionedTables;
 
     /** Creates a new {@link PostgresSourceConfig} for the given subtask {@code subtaskId}. */

pom.xml

@@ -59,7 +59,7 @@ limitations under the License.
     </scm>
 
     <properties>
-        <revision>3.5-SNAPSHOT</revision>
+        <revision>3.6-SNAPSHOT</revision>
         <scala.binary.version>2.12</scala.binary.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <!-- Enforce single fork execution due to heavy mini cluster use in the tests -->
@@ -421,6 +421,8 @@ limitations under the License.
                     </licenseFamilies>
                     <excludes>
                         <!-- Additional files like .gitignore etc.-->
+                        <!-- Release files -->
+                        <exclude>**/release/**</exclude>
                         <exclude>**/README.md</exclude>
                         <exclude>**/.*/**</exclude>
                         <!-- Generated content -->
@@ -676,6 +678,16 @@ limitations under the License.
                     <target>${maven.compiler.target}</target>
                 </configuration>
             </plugin>
+
+            <!-- Allow modification of pom.xml properties using Maven commands. -->
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>versions-maven-plugin</artifactId>
+                <version>2.5</version>
+                <configuration>
+                    <generateBackupPoms>false</generateBackupPoms>
+                </configuration>
+            </plugin>
         </plugins>
     </build>