[FLINK-32032] Upgrade to flink-shaded 17.0. This closes apache#22546

* Upgrade to flink-shaded 17.0
* Replace all `guava30` occurrences with `guava31` since flink-shaded uses a newer version of Guava
* Upgrade Curator to 5.4.0
* Update the flink-connector-jdbc as used by `flink-python` to 3.1.1-1.17 since that version doesn't rely on Flink Shaded
* Use a specific `flink-shaded-guava` version in the Confluent Schema Registry test to avoid ClassNotFound errors

flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e

@@ -321,7 +321,7 @@ Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(or
 Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.YearMonthIntervalType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.intGenerator(int, int)> in (RandomGeneratorVisitor.java:292)
 Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.YearMonthIntervalType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:293)
 Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.ZonedTimestampType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:331)
-Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor$2.next()> calls method <org.apache.flink.shaded.guava30.com.google.common.primitives.Longs.toByteArray(long)> in (SequenceGeneratorVisitor.java:221)
+Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor$2.next()> calls method <org.apache.flink.shaded.guava31.com.google.common.primitives.Longs.toByteArray(long)> in (SequenceGeneratorVisitor.java:221)
 Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor.getSequenceBytesGenerator(long, long)> has return type <org.apache.flink.streaming.api.functions.source.datagen.SequenceGenerator> in (SequenceGeneratorVisitor.java:0)
 Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor.getSequenceStringGenerator(long, long)> has return type <org.apache.flink.streaming.api.functions.source.datagen.SequenceGenerator> in (SequenceGeneratorVisitor.java:0)
 Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor.visit(org.apache.flink.table.types.logical.BigIntType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.SequenceGenerator.longGenerator(long, long)> in (SequenceGeneratorVisitor.java:172)
@@ -825,8 +825,8 @@ Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.addColumn(java.la
 Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.addColumn(java.lang.String, java.lang.String, org.apache.flink.table.types.DataType)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (HBaseTableSchema.java:73)
 Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.addColumn(java.lang.String, java.lang.String, org.apache.flink.table.types.DataType)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (HBaseTableSchema.java:74)
 Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.addColumn(java.lang.String, java.lang.String, org.apache.flink.table.types.DataType)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (HBaseTableSchema.java:75)
-Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.convertToDataType()> calls method <org.apache.flink.shaded.guava30.com.google.common.collect.Streams.zip(java.util.stream.Stream, java.util.stream.Stream, java.util.function.BiFunction)> in (HBaseTableSchema.java:283)
-Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.convertToDataType()> calls method <org.apache.flink.shaded.guava30.com.google.common.collect.Streams.zip(java.util.stream.Stream, java.util.stream.Stream, java.util.function.BiFunction)> in (HBaseTableSchema.java:298)
+Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.convertToDataType()> calls method <org.apache.flink.shaded.guava31.com.google.common.collect.Streams.zip(java.util.stream.Stream, java.util.stream.Stream, java.util.function.BiFunction)> in (HBaseTableSchema.java:283)
+Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.convertToDataType()> calls method <org.apache.flink.shaded.guava31.com.google.common.collect.Streams.zip(java.util.stream.Stream, java.util.stream.Stream, java.util.function.BiFunction)> in (HBaseTableSchema.java:298)
 Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.fromDataType(org.apache.flink.table.types.DataType)> calls method <org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType(org.apache.flink.table.types.logical.LogicalType)> in (HBaseTableSchema.java:335)
 Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.fromDataType(org.apache.flink.table.types.DataType)> calls method <org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType(org.apache.flink.table.types.logical.LogicalType)> in (HBaseTableSchema.java:338)
 Method <org.apache.flink.connector.hbase.util.HBaseTableSchema.getQualifierDataTypes(java.lang.String)> has return type <[Lorg.apache.flink.table.types.DataType;> in (HBaseTableSchema.java:0)

flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java

@@ -40,8 +40,8 @@
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.ShutdownHookUtil;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.MapDifference;
-import org.apache.flink.shaded.guava30.com.google.common.collect.Maps;
+import org.apache.flink.shaded.guava31.com.google.common.collect.MapDifference;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Maps;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

flink-clients/src/test/java/org/apache/flink/client/deployment/application/JarManifestParserTest.java

@@ -21,8 +21,8 @@
 import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.client.testjar.TestJob;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;

flink-clients/src/test/java/org/apache/flink/client/program/PerJobMiniClusterFactoryTest.java

@@ -32,7 +32,7 @@
 import org.apache.flink.runtime.testutils.CancelableInvokable;
 import org.apache.flink.runtime.testutils.WaitingCancelableInvokable;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;

flink-clients/src/test/java/org/apache/flink/client/testjar/ForbidConfigurationJob.java

@@ -23,7 +23,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
 
 /** Simulate a class setting a configuration. */
 public class ForbidConfigurationJob {

flink-connectors/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/HBaseConnectorITCase.java

@@ -40,7 +40,7 @@
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.CollectionUtil;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
 
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.util.Bytes;

flink-connectors/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/source/HBaseRowDataAsyncLookupFunctionTest.java

@@ -25,7 +25,7 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
 
 import org.junit.Test;
 

flink-connectors/flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/util/HBaseTableSchema.java

@@ -29,7 +29,7 @@
 import org.apache.flink.table.types.utils.TypeConversions;
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Streams;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Streams;
 
 import java.io.Serializable;
 import java.nio.charset.Charset;

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserCalcitePlanner.java

@@ -57,7 +57,7 @@
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
 
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;

flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkEmbeddedHiveRunner.java

@@ -20,7 +20,7 @@
 
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava30.com.google.common.io.Resources;
+import org.apache.flink.shaded.guava31.com.google.common.io.Resources;
 
 import com.klarna.hiverunner.HiveServerContainer;
 import com.klarna.hiverunner.HiveServerContext;

flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/FlinkStandaloneHiveRunner.java

@@ -20,8 +20,8 @@
 
 import org.apache.flink.table.catalog.hive.HiveTestUtils;
 
-import org.apache.flink.shaded.guava30.com.google.common.base.Joiner;
-import org.apache.flink.shaded.guava30.com.google.common.base.Throwables;
+import org.apache.flink.shaded.guava31.com.google.common.base.Joiner;
+import org.apache.flink.shaded.guava31.com.google.common.base.Throwables;
 
 import com.klarna.hiverunner.HiveServerContext;
 import org.apache.hadoop.hive.conf.HiveConf;

flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java

@@ -43,7 +43,7 @@
 import org.apache.flink.util.CollectionUtil;
 import org.apache.flink.util.TestLoggerExtension;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
 
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.junit.jupiter.api.AfterAll;

flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogITCase.java

@@ -46,7 +46,7 @@
 import org.apache.flink.util.CollectionUtil;
 import org.apache.flink.util.FileUtils;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;

flink-core/src/main/java/org/apache/flink/api/common/io/OutputFormatBase.java

@@ -22,7 +22,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.FutureCallback;
+import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.FutureCallback;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

flink-core/src/main/java/org/apache/flink/api/common/io/ratelimiting/GuavaFlinkConnectorRateLimiter.java

@@ -20,7 +20,7 @@
 
 import org.apache.flink.api.common.functions.RuntimeContext;
 
-import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.RateLimiter;
+import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.RateLimiter;
 
 /**
  * An implementation of {@link FlinkConnectorRateLimiter} that uses Guava's RateLimiter for rate

flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/GuavaRateLimiter.java

@@ -21,7 +21,7 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
-import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.RateLimiter;
+import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.RateLimiter;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;

flink-core/src/main/java/org/apache/flink/configuration/CleanupOptions.java

@@ -25,7 +25,7 @@
 import org.apache.flink.configuration.description.Description;
 import org.apache.flink.configuration.description.TextElement;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableSet;
 
 import java.time.Duration;
 import java.util.Set;

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -26,7 +26,7 @@
 import org.apache.flink.configuration.description.Description;
 import org.apache.flink.util.ArrayUtils;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
 
 import java.util.List;
 

flink-core/src/main/java/org/apache/flink/configuration/description/TextElement.java

@@ -20,7 +20,7 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 
-import org.apache.flink.shaded.guava30.com.google.common.base.Strings;
+import org.apache.flink.shaded.guava31.com.google.common.base.Strings;
 
 import java.util.Arrays;
 import java.util.Collections;

flink-core/src/main/java/org/apache/flink/core/classloading/ComponentClassLoader.java

@@ -20,7 +20,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.util.function.FunctionWithException;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Iterators;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Iterators;
 
 import java.io.IOException;
 import java.io.InputStream;

flink-core/src/main/java/org/apache/flink/core/fs/AutoCloseableRegistry.java

@@ -30,7 +30,7 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.flink.shaded.guava30.com.google.common.collect.Lists.reverse;
+import static org.apache.flink.shaded.guava31.com.google.common.collect.Lists.reverse;
 
 /**
  * This class allows to register instances of {@link AutoCloseable}, which are all closed if this

flink-core/src/main/java/org/apache/flink/core/fs/CloseableRegistry.java

@@ -30,7 +30,7 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.flink.shaded.guava30.com.google.common.collect.Lists.reverse;
+import static org.apache.flink.shaded.guava31.com.google.common.collect.Lists.reverse;
 
 /**
  * This class allows to register instances of {@link Closeable}, which are all closed if this

flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java

@@ -35,10 +35,10 @@
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.TemporaryClassLoaderContext;
 
-import org.apache.flink.shaded.guava30.com.google.common.base.Splitter;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMultimap;
-import org.apache.flink.shaded.guava30.com.google.common.collect.Iterators;
-import org.apache.flink.shaded.guava30.com.google.common.collect.Multimap;
+import org.apache.flink.shaded.guava31.com.google.common.base.Splitter;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMultimap;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Iterators;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Multimap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

flink-core/src/main/java/org/apache/flink/core/plugin/DefaultPluginManager.java

@@ -21,8 +21,8 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
 
-import org.apache.flink.shaded.guava30.com.google.common.base.Joiner;
-import org.apache.flink.shaded.guava30.com.google.common.collect.Iterators;
+import org.apache.flink.shaded.guava31.com.google.common.base.Joiner;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Iterators;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

flink-core/src/main/java/org/apache/flink/util/FileUtils.java

@@ -628,7 +628,7 @@ public static java.nio.file.Path getCurrentWorkingDirectory() {
      */
     public static boolean isJarFile(java.nio.file.Path file) {
         return JAR_FILE_EXTENSION.equals(
-                org.apache.flink.shaded.guava30.com.google.common.io.Files.getFileExtension(
+                org.apache.flink.shaded.guava31.com.google.common.io.Files.getFileExtension(
                         file.toString()));
     }
 
@@ -640,7 +640,7 @@ public static boolean isJarFile(java.nio.file.Path file) {
      */
     public static String stripFileExtension(String fileName) {
         final String extension =
-                org.apache.flink.shaded.guava30.com.google.common.io.Files.getFileExtension(
+                org.apache.flink.shaded.guava31.com.google.common.io.Files.getFileExtension(
                         fileName);
         if (!extension.isEmpty()) {
             return fileName.substring(0, fileName.lastIndexOf(extension) - 1);

flink-core/src/main/java/org/apache/flink/util/NetUtils.java

@@ -21,7 +21,7 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.configuration.IllegalConfigurationException;
 
-import org.apache.flink.shaded.guava30.com.google.common.net.InetAddresses;
+import org.apache.flink.shaded.guava31.com.google.common.net.InetAddresses;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

flink-core/src/test/java/org/apache/flink/configuration/ConfigOptionTest.java

@@ -20,7 +20,7 @@
 
 import org.apache.flink.util.TestLogger;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Sets;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Sets;
 
 import org.junit.Test;
 

flink-core/src/test/java/org/apache/flink/core/io/SimpleVersionedSerializationTest.java

@@ -21,7 +21,7 @@
 import org.apache.flink.core.memory.DataInputDeserializer;
 import org.apache.flink.core.memory.DataOutputSerializer;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
 
 import org.junit.Test;
 

flink-core/src/test/java/org/apache/flink/core/plugin/PluginConfigTest.java

@@ -22,7 +22,7 @@
 import org.apache.flink.core.testutils.CommonTestUtils;
 import org.apache.flink.util.TestLogger;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
 
 import org.junit.After;
 import org.junit.BeforeClass;

flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java

@@ -23,7 +23,7 @@
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.runtime.util.bash.BashJavaUtils;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Sets;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Sets;
 
 import org.hamcrest.Matchers;
 import org.junit.Test;

flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/AbstractStateChangeFsUploader.java

@@ -25,7 +25,7 @@
 import org.apache.flink.util.clock.Clock;
 import org.apache.flink.util.clock.SystemClock;
 
-import org.apache.flink.shaded.guava30.com.google.common.io.Closer;
+import org.apache.flink.shaded.guava31.com.google.common.io.Closer;
 
 import java.io.IOException;
 import java.util.Collection;

flink-dstl/flink-dstl-dfs/src/test/java/org/apache/flink/changelog/fs/FsStateChangelogWriterTest.java

@@ -37,7 +37,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
-import static org.apache.flink.shaded.guava30.com.google.common.collect.Iterables.getOnlyElement;
+import static org.apache.flink.shaded.guava31.com.google.common.collect.Iterables.getOnlyElement;
 import static org.apache.flink.util.ExceptionUtils.rethrowIOException;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;

flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml

@@ -53,6 +53,13 @@ under the License.
 			<artifactId>flink-connector-kafka</artifactId>
 			<version>3.0.0-1.17</version>
 		</dependency>
+		<!-- Make sure that Shaded Guava matches the one used in the flink-connector-kafka,
+		 or remove when FLINK-32462 is resolved -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-shaded-guava</artifactId>
+			<version>30.1.1-jre-16.1</version>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-avro</artifactId>

flink-end-to-end-tests/flink-queryable-state-test/src/main/java/org/apache/flink/streaming/tests/queryablestate/QsStateProducer.java

@@ -37,7 +37,7 @@
 import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
 import org.apache.flink.util.Collector;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
 
 import java.time.Duration;
 import java.time.Instant;

flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/utils/ChecksumUtils.java

@@ -18,8 +18,8 @@
 
 package org.apache.flink.fs.gs.utils;
 
-import org.apache.flink.shaded.guava30.com.google.common.hash.HashFunction;
-import org.apache.flink.shaded.guava30.com.google.common.hash.Hashing;
+import org.apache.flink.shaded.guava31.com.google.common.hash.HashFunction;
+import org.apache.flink.shaded.guava31.com.google.common.hash.Hashing;
 
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;

flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/writer/GSChecksumWriteChannel.java

@@ -23,7 +23,7 @@
 import org.apache.flink.fs.gs.utils.ChecksumUtils;
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava30.com.google.common.hash.Hasher;
+import org.apache.flink.shaded.guava31.com.google.common.hash.Hasher;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

flink-filesystems/flink-gs-fs-hadoop/src/test/java/org/apache/flink/fs/gs/utils/ConfigUtilsHadoopTest.java

@@ -21,8 +21,8 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.fs.gs.TestUtils;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.MapDifference;
-import org.apache.flink.shaded.guava30.com.google.common.collect.Maps;
+import org.apache.flink.shaded.guava31.com.google.common.collect.MapDifference;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Maps;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;

flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAJobRunOnMinioS3StoreITCase.java

@@ -30,7 +30,7 @@
 import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.test.junit5.MiniClusterExtension;
 
-import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
+import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
 
 import com.amazonaws.services.s3.model.S3ObjectSummary;
 import org.apache.commons.lang3.StringUtils;
@@ -40,7 +40,7 @@
 
 import java.util.List;
 
-import static org.apache.flink.shaded.guava30.com.google.common.base.Predicates.not;
+import static org.apache.flink.shaded.guava31.com.google.common.base.Predicates.not;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**