Skip to content

[FLINK-39152][pipeline-connector][Fluss]Fix FlussConfig Utils class is not in the correct directory and not have a private instance #4285

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ThorneANN wants to merge 16 commits into from
Closed

[FLINK-39152][pipeline-connector][Fluss]Fix FlussConfig Utils class is not in the correct directory and not have a private instance #4285

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
dea3678
Fix FlussConfig Utils class is not in the correct directory and does …
ThorneANN Feb 25, 2026
53a78c9
fix some normal code
ThorneANN Feb 25, 2026
524d9d6
fix done
ThorneANN Feb 25, 2026
8691a08
change the log info value
ThorneANN Feb 25, 2026
6f052fb
[FLINK-39001][doc][Flink-source]supple NewlyAddTable's doc with mongo…
ThorneANN Feb 25, 2026
5b56a5f
[pipeline] Use GHA workflow triggers to run labeler workflow (#4274)
yuxiqian Feb 26, 2026
1bc4050
[FLINK-39071] Select fields rather * to read snapshot data in case sc…
loserwang1024 Feb 27, 2026
6eade69
[Minor][FLINK-39168][cdc-connector-mysql] Cache split.splitId() resul…
ThorneANN Feb 27, 2026
094125c
[ci] Fix docker environment related test failure (#4275)
yuxiqian Feb 27, 2026
5b2dff8
[FLINK-37485][starrocks] Add support for TIME type (#4253)
Hisoka-X Mar 3, 2026
0bbedd5
[chore][test] Fix flaky postgres pipeline test case (#4293)
Hisoka-X Mar 4, 2026
1048461
[FLINK-38833][paimon] Shuffle record to different subtasks by table, …
sd4324530 Mar 4, 2026
997eb15
[FLINK-38334][mysql] Fix MySQL CDC source stuck in INITIAL_ASSIGNING …
morozov Mar 5, 2026
86424d3
[FLINK-37586][udf] Add support for options in user-defined functions …
Hisoka-X Mar 5, 2026
45cb0e3
[FLINK-38726][fluss] Bump Fluss version to 0.9.0-incubating
leonardBang Mar 6, 2026
8ff1ee4
suite fluss 0.9 verison
ThorneANN Mar 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ...uss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,13 +32,13 @@
import java.util.Map;
import java.util.Set;

import static org.apache.flink.cdc.connectors.fluss.sink.FlussConfigUtils.parseBucketKeys;
import static org.apache.flink.cdc.connectors.fluss.sink.FlussConfigUtils.parseBucketNumber;
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BOOTSTRAP_SERVERS;
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_KEY;
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_NUMBER;
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.CLIENT_PROPERTIES_PREFIX;
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.TABLE_PROPERTIES_PREFIX;
import static org.apache.flink.cdc.connectors.fluss.utils.FlussConfigUtils.parseBucketKeys;
import static org.apache.flink.cdc.connectors.fluss.utils.FlussConfigUtils.parseBucketNumber;

/** Factory for creating configured instances of {@link FlussDataSink}. */
public class FlussDataSinkFactory implements DataSinkFactory {
Expand Down
15 changes: 7 additions & 8 deletions ...nnectors/fluss/sink/FlussConfigUtils.java → ...nectors/fluss/utils/FlussConfigUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.flink.cdc.connectors.fluss.sink;
package org.apache.flink.cdc.connectors.fluss.utils;

import javax.annotation.Nullable;

Expand All @@ -24,7 +24,7 @@
import java.util.List;
import java.util.Map;

/** Utils for parsing fluss yaml sink options. */
/** Utils for parsing fluss Yaml sink options. */
public class FlussConfigUtils {
public static Map<String, List<String>> parseBucketKeys(@Nullable String rawValue)
throws IllegalArgumentException {
Expand Down Expand Up @@ -64,18 +64,17 @@ public static Map<String, Integer> parseBucketNumber(@Nullable String rawValue)
String[] kv = tableBucketNumStr.trim().split(":", 2);
if (kv.length != 2) {
throw new IllegalArgumentException(
"Invalid bucket number configuration: " + rawValue);
"Invalid bucket number configuration for : " + rawValue);
}

String table = kv[0].trim();
try {
int value = Integer.parseInt(kv[1].trim());
result.put(table, value);
result.put(kv[0].trim(), Integer.valueOf(kv[1].trim()));
} catch (NumberFormatException ignored) {
throw new IllegalArgumentException(
"Invalid bucket number configuration: " + rawValue);
"Invalid bucket number configuration for : " + rawValue);
}
}
return result;
}

private FlussConfigUtils() {}
}
3 changes: 2 additions & 1 deletion ...src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,8 @@ void testWrongBucketKeyAndBucketNum() {
.build()),
new Configuration(),
Thread.currentThread().getContextClassLoader())))
.hasMessageContaining("Invalid bucket number configuration: database1.table1: 11a");
.hasMessageContaining(
"Invalid bucket number configuration for : database1.table1: 11a");
}

private Configuration createValidConfiguration() {
Expand Down
Loading