From 5abd91d1431267dca94c39cdf23fe170db9212fa Mon Sep 17 00:00:00 2001 From: Jayant Jain Date: Mon, 27 Jan 2025 19:44:28 +0530 Subject: [PATCH] Handle case mismatch in region input --- .../google/cloud/flink/bigquery/sink/BigQueryBaseSink.java | 6 ++++-- .../cloud/flink/bigquery/sink/BigQueryDefaultSinkTest.java | 4 ++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/flink-1.17-connector-bigquery/flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/sink/BigQueryBaseSink.java b/flink-1.17-connector-bigquery/flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/sink/BigQueryBaseSink.java index fdea78e8..aabe0a3d 100644 --- a/flink-1.17-connector-bigquery/flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/sink/BigQueryBaseSink.java +++ b/flink-1.17-connector-bigquery/flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/sink/BigQueryBaseSink.java @@ -151,8 +151,10 @@ private String getRegion(String userProvidedRegion) { // Max sink parallelism is deduced using destination dataset's region. // Ensure instance variable 'region' is assigned before invoking this method. private int getMaxParallelism() { - if (BQ_MULTI_REGIONS.contains(region)) { - return MULTI_REGION_MAX_SINK_PARALLELISM; + for (String multiRegion : BQ_MULTI_REGIONS) { + if (multiRegion.equalsIgnoreCase(region)) { + return MULTI_REGION_MAX_SINK_PARALLELISM; + } } return DEFAULT_MAX_SINK_PARALLELISM; } diff --git a/flink-1.17-connector-bigquery/flink-connector-bigquery/src/test/java/com/google/cloud/flink/bigquery/sink/BigQueryDefaultSinkTest.java b/flink-1.17-connector-bigquery/flink-connector-bigquery/src/test/java/com/google/cloud/flink/bigquery/sink/BigQueryDefaultSinkTest.java index 265a3905..2a00541f 100644 --- a/flink-1.17-connector-bigquery/flink-connector-bigquery/src/test/java/com/google/cloud/flink/bigquery/sink/BigQueryDefaultSinkTest.java +++ b/flink-1.17-connector-bigquery/flink-connector-bigquery/src/test/java/com/google/cloud/flink/bigquery/sink/BigQueryDefaultSinkTest.java @@ -262,10 +262,10 @@ public void testRegionAndParallelism_useProvidedMultiRegion() { .serializer(new AvroToProtoSerializer()) .enableTableCreation(true) .streamExecutionEnvironment(env) - .region("us") + .region("US") .build(); BigQueryDefaultSink sink = new BigQueryDefaultSink(sinkConfig); - assertEquals("us", sink.region); + assertEquals("US", sink.region); assertEquals(512, sink.maxParallelism); }