Adding configurable option to set source parallelism by KevBrowne · Pull Request #103 · pinterest/psc

KevBrowne · 2025-10-21T20:23:12Z

Summary

Currently, PscDynamicSource defaults to parallelism equal to the number of topic partitions. Under Flink's FLIP-27 Source API, when parallelism is not explicitly set, Flink automatically derives it from the number of splits (topic partitions in this case). This creates bottlenecks when downstream operators need higher parallelism, as the source operator cannot scale beyond the partition count, leading to:

Underutilized task slots in downstream operators
Performance bottlenecks despite available resources
Inefficient resource allocation in large-scale pipelines

Solution

Added scan.enable-rescale** configuration option that determines when data redistribution is needed. When enabled, the connector:

Queries actual partition count from PSC metadata service
Compares with configured parallelism (table.exec.resource.default-parallelism) - https://github.com/apache/flink/blob/master/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java#L314
Automatically applies rescale() ONLY when parallelism > partition count
Avoids unnecessary shuffle when partitions >= parallelism (no performance cost)

This eliminates manual calculation and automatically optimizes based on actual cluster state.

Added single boolean configuration option:

scan.enable-rescale (Boolean, default: false)
- Enables rescale() shuffle to redistribute data
- Default false to avoid unnecessary overhead when partitions >= pipeline parallelism
- Safe to enable when topic partitions < downstream operator parallelism

Configure Pipeline Parallelism

# flink-conf.yaml or configured properties files
table.exec.resource.default-parallelism: 10

Or via SQL:

SET 'table.exec.resource.default-parallelism' = '10';

Usage

Scenario 1: Rescale Applied

Topic has 2 partitions
Configured parallelism = 10
Result: Connector detects 10 > 2, applies rescale() automatically
Log: "Applying rescale(): configured parallelism (10) > partition count (2)"

Scenario 2: Rescale Skipped

Topic has 8 partitions
Configured parallelism = 4
Result: Connector detects 4 ≤ 8, skips rescale() automatically
Log: "Skipping rescale(): configured parallelism (4) <= partition count (8)"

When to enable:

Topic has fewer partitions than pipeline parallelism
Downstream operators need higher parallelism than partition count
Performance bottleneck is in downstream processing, not source reading

When to keep disabled (default):

Topic partitions >= downstream operator parallelism
Want to preserve per-partition ordering
Want to minimize network shuffle overhead

psc % mvn -pl psc-flink -am clean compile -DskipTests -Dmaven.javadoc.skip=true -Dgpg.skip
WARNING: A terminally deprecated method in sun.misc.Unsafe has been called
WARNING: sun.misc.Unsafe::staticFieldBase has been called by com.google.inject.internal.aop.HiddenClassDefiner (file:/opt/homebrew/Cellar/maven/3.9.11/libexec/lib/guice-5.1.0-classes.jar)
WARNING: Please consider reporting this to the maintainers of class com.google.inject.internal.aop.HiddenClassDefiner
WARNING: sun.misc.Unsafe::staticFieldBase will be removed in a future release
[INFO] Scanning for projects...
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-examples:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-integration-test:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-flink:jar:4.1.3-SNAPSHOT
[WARNING] 'dependencies.dependency.(groupId:artifactId:type:classifier)' must be unique: org.apache.flink:flink-table-planner_${scala.binary.version}:jar -> duplicate declaration of version ${flink.version} @ line 327, column 21
[WARNING] 'dependencies.dependency.(groupId:artifactId:type:classifier)' must be unique: org.apache.flink:flink-json:jar -> duplicate declaration of version ${flink.version} @ line 417, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-logging:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-common:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-flink-logging:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-java-oss:pom:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ line 160, column 21
[WARNING] 
[WARNING] It is highly recommended to fix these problems because they threaten the stability of your build.
[WARNING] 
[WARNING] For this reason, future Maven versions might no longer support building such malformed projects.
[WARNING] 
[INFO] Inspecting build with total of 6 modules...
[INFO] Installing Nexus Staging features:
[INFO]   ... total of 6 executions of maven-deploy-plugin replaced with nexus-staging-maven-plugin
[INFO] ------------------------------------------------------------------------
[INFO] Reactor Build Order:
[INFO] 
[INFO] psc-java-oss                                                       [pom]
[INFO] psc-common                                                         [jar]
[INFO] psc-logging                                                        [jar]
[INFO] psc                                                                [jar]
[INFO] psc-flink-logging                                                  [jar]
[INFO] psc-flink                                                          [jar]
[INFO] 
[INFO] -------------------< com.pinterest.psc:psc-java-oss >-------------------
[INFO] Building psc-java-oss 4.1.3-SNAPSHOT                               [1/6]
[INFO]   from pom.xml
[INFO] --------------------------------[ pom ]---------------------------------
[WARNING] Ignoring incompatible plugin version 4.0.0-beta-1: The plugin org.apache.maven.plugins:maven-source-plugin:4.0.0-beta-1 requires Maven version 4.0.0-beta-3
[INFO] Latest version of plugin org.apache.maven.plugins:maven-source-plugin failed compatibility check
[INFO] Looking for compatible RELEASE version of plugin org.apache.maven.plugins:maven-source-plugin
[INFO] Selected plugin org.apache.maven.plugins:maven-source-plugin:3.3.1
[INFO] 
[INFO] --- clean:3.2.0:clean (default-clean) @ psc-java-oss ---
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc-java-oss ---
[INFO] argLine set to -javaagent:/Users/kbrowne/.m2/repository/org/jacoco/org.jacoco.agent/0.8.5/org.jacoco.agent-0.8.5-runtime.jar=destfile=/Users/kbrowne/code/psc/target/jacoco.exec
[INFO] 
[INFO] --------------------< com.pinterest.psc:psc-common >--------------------
[INFO] Building psc-common 4.1.3-SNAPSHOT                                 [2/6]
[INFO]   from psc-common/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- clean:3.2.0:clean (default-clean) @ psc-common ---
[INFO] Deleting /Users/kbrowne/code/psc/psc-common/target
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc-common ---
[INFO] argLine set to -javaagent:/Users/kbrowne/.m2/repository/org/jacoco/org.jacoco.agent/0.8.5/org.jacoco.agent-0.8.5-runtime.jar=destfile=/Users/kbrowne/code/psc/psc-common/target/jacoco.exec
[INFO] 
[INFO] --- resources:3.3.1:resources (default-resources) @ psc-common ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] skip non existing resourceDirectory /Users/kbrowne/code/psc/psc-common/src/main/resources
[INFO] 
[INFO] --- compiler:3.8.1:compile (default-compile) @ psc-common ---
[INFO] Changes detected - recompiling the module!
[WARNING] File encoding has not been set, using platform encoding UTF-8, i.e. build is platform dependent!
[INFO] Compiling 2 source files to /Users/kbrowne/code/psc/psc-common/target/classes
[INFO] /Users/kbrowne/code/psc/psc-common/src/main/java/com/pinterest/psc/common/PscCommon.java: /Users/kbrowne/code/psc/psc-common/src/main/java/com/pinterest/psc/common/PscCommon.java uses unchecked or unsafe operations.
[INFO] /Users/kbrowne/code/psc/psc-common/src/main/java/com/pinterest/psc/common/PscCommon.java: Recompile with -Xlint:unchecked for details.
[INFO] 
[INFO] -------------------< com.pinterest.psc:psc-logging >--------------------
[INFO] Building psc-logging 4.1.3-SNAPSHOT                                [3/6]
[INFO]   from psc-logging/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- clean:3.2.0:clean (default-clean) @ psc-logging ---
[INFO] Deleting /Users/kbrowne/code/psc/psc-logging/target
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc-logging ---
[INFO] argLine set to -javaagent:/Users/kbrowne/.m2/repository/org/jacoco/org.jacoco.agent/0.8.5/org.jacoco.agent-0.8.5-runtime.jar=destfile=/Users/kbrowne/code/psc/psc-logging/target/jacoco.exec
[INFO] 
[INFO] --- resources:3.3.1:resources (default-resources) @ psc-logging ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] skip non existing resourceDirectory /Users/kbrowne/code/psc/psc-logging/src/main/resources
[INFO] 
[INFO] --- compiler:3.8.1:compile (default-compile) @ psc-logging ---
[INFO] Changes detected - recompiling the module!
[WARNING] File encoding has not been set, using platform encoding UTF-8, i.e. build is platform dependent!
[INFO] Compiling 5 source files to /Users/kbrowne/code/psc/psc-logging/target/classes
[INFO] 
[INFO] -----------------------< com.pinterest.psc:psc >------------------------
[INFO] Building psc 4.1.3-SNAPSHOT                                        [4/6]
[INFO]   from psc/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[WARNING] 1 problem was encountered while building the effective model for org.javassist:javassist:jar:3.18.2-GA during dependency collection step for project (use -X to see details)
[INFO] 
[INFO] --- clean:3.2.0:clean (default-clean) @ psc ---
[INFO] Deleting /Users/kbrowne/code/psc/psc/target
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc ---
[INFO] argLine set to -javaagent:/Users/kbrowne/.m2/repository/org/jacoco/org.jacoco.agent/0.8.5/org.jacoco.agent-0.8.5-runtime.jar=destfile=/Users/kbrowne/code/psc/psc/target/jacoco.exec
[INFO] 
[INFO] --- resources:3.3.1:resources (default-resources) @ psc ---
[WARNING] File encoding has not been set, using platform encoding UTF-8. Build is platform dependent!
[WARNING] See https://maven.apache.org/general.html#encoding-warning
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 2 resources from src/main/resources to target/classes
[INFO] Copying 1 resource from src/main/resources-filtered to target/classes
[INFO] The encoding used to copy filtered properties files have not been set. This means that the same encoding will be used to copy filtered properties files as when copying other filtered resources. This might not be what you want! Run your build with --debug to see which files might be affected. Read more at https://maven.apache.org/plugins/maven-resources-plugin/examples/filtering-properties-files.html
[INFO] 
[INFO] --- compiler:3.8.1:compile (default-compile) @ psc ---
[INFO] Changes detected - recompiling the module!
[WARNING] File encoding has not been set, using platform encoding UTF-8, i.e. build is platform dependent!
[INFO] Compiling 162 source files to /Users/kbrowne/code/psc/psc/target/classes
[INFO] /Users/kbrowne/code/psc/psc/src/main/java/com/pinterest/psc/common/kafka/KafkaErrors.java: Some input files use or override a deprecated API.
[INFO] /Users/kbrowne/code/psc/psc/src/main/java/com/pinterest/psc/common/kafka/KafkaErrors.java: Recompile with -Xlint:deprecation for details.
[INFO] /Users/kbrowne/code/psc/psc/src/main/java/com/pinterest/psc/config/PscConfigurationInternal.java: Some input files use unchecked or unsafe operations.
[INFO] /Users/kbrowne/code/psc/psc/src/main/java/com/pinterest/psc/config/PscConfigurationInternal.java: Recompile with -Xlint:unchecked for details.
[INFO] 
[INFO] ----------------< com.pinterest.psc:psc-flink-logging >-----------------
[INFO] Building psc-flink-logging 4.1.3-SNAPSHOT                          [5/6]
[INFO]   from psc-flink-logging/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- clean:3.2.0:clean (default-clean) @ psc-flink-logging ---
[INFO] Deleting /Users/kbrowne/code/psc/psc-flink-logging/target
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc-flink-logging ---
[INFO] argLine set to -javaagent:/Users/kbrowne/.m2/repository/org/jacoco/org.jacoco.agent/0.8.5/org.jacoco.agent-0.8.5-runtime.jar=destfile=/Users/kbrowne/code/psc/psc-flink-logging/target/jacoco.exec
[INFO] 
[INFO] --- resources:3.3.1:resources (default-resources) @ psc-flink-logging ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] skip non existing resourceDirectory /Users/kbrowne/code/psc/psc-flink-logging/src/main/resources
[INFO] 
[INFO] --- compiler:3.8.1:compile (default-compile) @ psc-flink-logging ---
[INFO] Changes detected - recompiling the module!
[WARNING] File encoding has not been set, using platform encoding UTF-8, i.e. build is platform dependent!
[INFO] Compiling 4 source files to /Users/kbrowne/code/psc/psc-flink-logging/target/classes
[INFO] 
[INFO] --------------------< com.pinterest.psc:psc-flink >---------------------
[INFO] Building psc-flink 4.1.3-SNAPSHOT                                  [6/6]
[INFO]   from psc-flink/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[WARNING] 2 problems were encountered while building the effective model for org.apache.yetus:audience-annotations:jar:0.5.0 during dependency collection step for project (use -X to see details)
[WARNING] 1 problem was encountered while building the effective model for org.javassist:javassist:jar:3.18.1-GA during dependency collection step for project (use -X to see details)
[INFO] 
[INFO] --- clean:3.2.0:clean (default-clean) @ psc-flink ---
[INFO] Deleting /Users/kbrowne/code/psc/psc-flink/target
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc-flink ---
[INFO] argLine set to -javaagent:/Users/kbrowne/.m2/repository/org/jacoco/org.jacoco.agent/0.8.5/org.jacoco.agent-0.8.5-runtime.jar=destfile=/Users/kbrowne/code/psc/psc-flink/target/jacoco.exec
[INFO] 
[INFO] --- resources:3.3.1:resources (default-resources) @ psc-flink ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 1 resource from src/main/resources to target/classes
[INFO] 
[INFO] --- compiler:3.8.1:compile (default-compile) @ psc-flink ---
[INFO] Changes detected - recompiling the module!
[WARNING] File encoding has not been set, using platform encoding UTF-8, i.e. build is platform dependent!
[INFO] Compiling 142 source files to /Users/kbrowne/code/psc/psc-flink/target/classes
[INFO] /Users/kbrowne/code/psc/psc-flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/internals/SourceContextWatermarkOutputAdapter.java: Some input files use or override a deprecated API.
[INFO] /Users/kbrowne/code/psc/psc-flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/internals/SourceContextWatermarkOutputAdapter.java: Recompile with -Xlint:deprecation for details.
[INFO] /Users/kbrowne/code/psc/psc-flink/src/main/java/com/pinterest/flink/connector/psc/dynamic/source/DynamicPscSourceOptions.java: Some input files use unchecked or unsafe operations.
[INFO] /Users/kbrowne/code/psc/psc-flink/src/main/java/com/pinterest/flink/connector/psc/dynamic/source/DynamicPscSourceOptions.java: Recompile with -Xlint:unchecked for details.
[INFO] ------------------------------------------------------------------------
[INFO] Reactor Summary for psc-java-oss 4.1.3-SNAPSHOT:
[INFO] 
[INFO] psc-java-oss ....................................... SUCCESS [  0.207 s]
[INFO] psc-common ......................................... SUCCESS [  0.341 s]
[INFO] psc-logging ........................................ SUCCESS [  0.093 s]
[INFO] psc ................................................ SUCCESS [  0.915 s]
[INFO] psc-flink-logging .................................. SUCCESS [  0.026 s]
[INFO] psc-flink .......................................... SUCCESS [  0.797 s]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  2.619 s
[INFO] Finished at: 2025-10-28T11:05:59-04:00
[INFO] ------------------------------------------------------------------------

Unit test

psc % mvn test -pl psc-flink test -Dtest=PscDynamicTableFactoryTest -Djacoco.skip=true
WARNING: A terminally deprecated method in sun.misc.Unsafe has been called
WARNING: sun.misc.Unsafe::staticFieldBase has been called by com.google.inject.internal.aop.HiddenClassDefiner (file:/opt/homebrew/Cellar/maven/3.9.11/libexec/lib/guice-5.1.0-classes.jar)
WARNING: Please consider reporting this to the maintainers of class com.google.inject.internal.aop.HiddenClassDefiner
WARNING: sun.misc.Unsafe::staticFieldBase will be removed in a future release
[INFO] Scanning for projects...
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-examples:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-integration-test:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-flink:jar:4.1.3-SNAPSHOT
[WARNING] 'dependencies.dependency.(groupId:artifactId:type:classifier)' must be unique: org.apache.flink:flink-table-planner_${scala.binary.version}:jar -> duplicate declaration of version ${flink.version} @ line 327, column 21
[WARNING] 'dependencies.dependency.(groupId:artifactId:type:classifier)' must be unique: org.apache.flink:flink-json:jar -> duplicate declaration of version ${flink.version} @ line 417, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-logging:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-common:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-flink-logging:jar:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ com.pinterest.psc:psc-java-oss:4.1.3-SNAPSHOT, /Users/kbrowne/code/psc/pom.xml, line 160, column 21
[WARNING] 
[WARNING] Some problems were encountered while building the effective model for com.pinterest.psc:psc-java-oss:pom:4.1.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for org.apache.maven.plugins:maven-source-plugin is missing. @ line 160, column 21
[WARNING] 
[WARNING] It is highly recommended to fix these problems because they threaten the stability of your build.
[WARNING] 
[WARNING] For this reason, future Maven versions might no longer support building such malformed projects.
[WARNING] 
[INFO] Inspecting build with total of 1 modules...
[INFO] Installing Nexus Staging features:
[INFO]   ... total of 1 executions of maven-deploy-plugin replaced with nexus-staging-maven-plugin
[INFO] 
[INFO] --------------------< com.pinterest.psc:psc-flink >---------------------
[INFO] Building psc-flink 4.1.3-SNAPSHOT
[INFO]   from pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[WARNING] 1 problem was encountered while building the effective model for org.javassist:javassist:jar:3.18.2-GA during dependency collection step for project (use -X to see details)
[WARNING] 2 problems were encountered while building the effective model for org.apache.yetus:audience-annotations:jar:0.5.0 during dependency collection step for project (use -X to see details)
[WARNING] 1 problem was encountered while building the effective model for org.javassist:javassist:jar:3.18.1-GA during dependency collection step for project (use -X to see details)
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc-flink ---
[INFO] Skipping JaCoCo execution because property jacoco.skip is set.
[INFO] argLine set to empty
[INFO] 
[INFO] --- resources:3.3.1:resources (default-resources) @ psc-flink ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 1 resource from src/main/resources to target/classes
[INFO] 
[INFO] --- compiler:3.8.1:compile (default-compile) @ psc-flink ---
[INFO] Nothing to compile - all classes are up to date
[INFO] 
[INFO] --- resources:3.3.1:testResources (default-testResources) @ psc-flink ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 99 resources from src/test/resources to target/test-classes
[INFO] 
[INFO] --- compiler:3.8.1:testCompile (default-testCompile) @ psc-flink ---
[INFO] Changes detected - recompiling the module!
[WARNING] File encoding has not been set, using platform encoding UTF-8, i.e. build is platform dependent!
[INFO] Compiling 113 source files to /Users/kbrowne/code/psc/psc-flink/target/test-classes
[WARNING] /Users/kbrowne/code/psc/psc-flink/src/test/java/com/pinterest/flink/connector/psc/sink/PscSinkITCase.java:[631,26] stop() in java.lang.Thread has been deprecated and marked for removal
[INFO] /Users/kbrowne/code/psc/psc-flink/src/test/java/com/pinterest/flink/streaming/connectors/psc/shuffle/PscShuffleITCase.java: Some input files use or override a deprecated API.
[INFO] /Users/kbrowne/code/psc/psc-flink/src/test/java/com/pinterest/flink/streaming/connectors/psc/shuffle/PscShuffleITCase.java: Recompile with -Xlint:deprecation for details.
[INFO] /Users/kbrowne/code/psc/psc-flink/src/test/java/com/pinterest/flink/connector/psc/dynamic/source/enumerator/DynamicPscSourceEnumeratorTest.java: Some input files use unchecked or unsafe operations.
[INFO] /Users/kbrowne/code/psc/psc-flink/src/test/java/com/pinterest/flink/connector/psc/dynamic/source/enumerator/DynamicPscSourceEnumeratorTest.java: Recompile with -Xlint:unchecked for details.
[INFO] 
[INFO] --- surefire:3.0.0-M5:test (default-test) @ psc-flink ---
[INFO] 
[INFO] -------------------------------------------------------
[INFO]  T E S T S
[INFO] -------------------------------------------------------
[INFO] Tests run: 56, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.362 s - in com.pinterest.flink.streaming.connectors.psc.table.PscDynamicTableFactoryTest
[INFO] 
[INFO] Results:
[INFO] 
[INFO] Tests run: 56, Failures: 0, Errors: 0, Skipped: 0
[INFO] 
[INFO] 
[INFO] --- jacoco:0.8.5:prepare-agent (prepare-unit-tests) @ psc-flink ---
[INFO] Skipping JaCoCo execution because property jacoco.skip is set.
[INFO] 
[INFO] --- resources:3.3.1:resources (default-resources) @ psc-flink ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 1 resource from src/main/resources to target/classes
[INFO] 
[INFO] --- compiler:3.8.1:compile (default-compile) @ psc-flink ---
[INFO] Nothing to compile - all classes are up to date
[INFO] 
[INFO] --- resources:3.3.1:testResources (default-testResources) @ psc-flink ---
[WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 99 resources from src/test/resources to target/test-classes
[INFO] 
[INFO] --- compiler:3.8.1:testCompile (default-testCompile) @ psc-flink ---
[INFO] Nothing to compile - all classes are up to date
[INFO] 
[INFO] --- surefire:3.0.0-M5:test (default-test) @ psc-flink ---
[INFO] Skipping execution of surefire because it has already been run for this configuration
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  1.706 s
[INFO] Finished at: 2025-10-28T11:14:55-04:00
[INFO] ------------------------------------------------------------------------

...nk/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscConnectorOptions.java

...flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscDynamicSource.java

...ink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscTableSourceUtil.java

...flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscDynamicSource.java

nickpan47

lgtm overall. Please refactor the dup code block. Thanks!

...in/java/com/pinterest/flink/streaming/connectors/psc/table/UpsertPscDynamicTableFactory.java

...test/java/com/pinterest/flink/streaming/connectors/psc/table/PscDynamicTableFactoryTest.java

...nk/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscTableCommonUtils.java

nickpan47

nit: suggest renaming the util class. Otherwise, lgtm. Thanks!

nickpan47

Ship it!

* Adding configurable optional to set source parallelism * fixes * make source infer parallelism * adding shuffle * revert pom * removing inference and keeping boolean flag * rework rescale apply logic * make common util * create TableCommonUtils --------- Co-authored-by: Kevin Browne <kbrowne@kbrowne-JG645XV.dyn.pinadmin.com>

* Adding configurable option to set source parallelism (#103) * Adding configurable optional to set source parallelism * fixes * make source infer parallelism * adding shuffle * revert pom * removing inference and keeping boolean flag * rework rescale apply logic * make common util * create TableCommonUtils --------- Co-authored-by: Kevin Browne <kbrowne@kbrowne-JG645XV.dyn.pinadmin.com> * Turn off error logging in flink tests --------- Co-authored-by: KevBrowne <kjbrowne0928@gmail.com> Co-authored-by: Kevin Browne <kbrowne@kbrowne-JG645XV.dyn.pinadmin.com>

Kevin Browne added 2 commits October 21, 2025 16:20

Adding configurable optional to set source parallelism

9dcc892

fixes

051e437

KevBrowne marked this pull request as ready for review October 21, 2025 21:02

KevBrowne requested a review from a team as a code owner October 21, 2025 21:02

KevBrowne changed the title ~~Adding configurable optional to set source parallelism~~ Adding configurable option to set source parallelism Oct 21, 2025

nickpan47 reviewed Oct 22, 2025

View reviewed changes

...nk/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscConnectorOptions.java Outdated Show resolved Hide resolved

make source infer parallelism

16708a9

KevBrowne requested a review from nickpan47 October 22, 2025 23:39

nickpan47 reviewed Oct 23, 2025

View reviewed changes

...flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscDynamicSource.java Outdated Show resolved Hide resolved

adding shuffle

a009168

KevBrowne requested a review from nickpan47 October 23, 2025 15:07

revert pom

2569d42

nickpan47 reviewed Oct 24, 2025

View reviewed changes

...flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscDynamicSource.java Outdated Show resolved Hide resolved

nickpan47 reviewed Oct 24, 2025

View reviewed changes

...ink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscTableSourceUtil.java Outdated Show resolved Hide resolved

removing inference and keeping boolean flag

40f5446

nickpan47 reviewed Oct 27, 2025

View reviewed changes

...flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscDynamicSource.java Show resolved Hide resolved

rework rescale apply logic

ace54aa

KevBrowne requested a review from nickpan47 October 28, 2025 15:24

nickpan47 suggested changes Oct 29, 2025

View reviewed changes

...in/java/com/pinterest/flink/streaming/connectors/psc/table/UpsertPscDynamicTableFactory.java Outdated Show resolved Hide resolved

...test/java/com/pinterest/flink/streaming/connectors/psc/table/PscDynamicTableFactoryTest.java Show resolved Hide resolved

make common util

89e91d4

KevBrowne requested a review from nickpan47 October 30, 2025 17:12

nickpan47 reviewed Oct 30, 2025

View reviewed changes

...nk/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscTableCommonUtils.java Show resolved Hide resolved

nickpan47 reviewed Oct 30, 2025

View reviewed changes

jeffxiang approved these changes Oct 31, 2025

View reviewed changes

create TableCommonUtils

e20b397

KevBrowne requested a review from nickpan47 November 3, 2025 16:19

nickpan47 approved these changes Nov 5, 2025

View reviewed changes

jeffxiang merged commit 55d6625 into pinterest:main Nov 5, 2025
2 of 3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding configurable option to set source parallelism#103

Adding configurable option to set source parallelism#103
jeffxiang merged 9 commits intopinterest:mainfrom
KevBrowne:add_configurable_source_parallelism

KevBrowne commented Oct 21, 2025 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

nickpan47 left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

nickpan47 left a comment

Uh oh!

nickpan47 left a comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

KevBrowne commented Oct 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Solution

Configure Pipeline Parallelism

Usage

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

KevBrowne commented Oct 21, 2025 •

edited

Loading