[FLINK-35038] Bump org.yaml:snakeyaml to 2.2 (#93)

* Bump org.yaml:snakeyaml from 1.31 to 2.0 in /flink-connector-kafka

Bumps [org.yaml:snakeyaml](https://bitbucket.org/snakeyaml/snakeyaml) from 1.31 to 2.0.
- [Commits](https://bitbucket.org/snakeyaml/snakeyaml/branches/compare/snakeyaml-2.0..snakeyaml-1.31)

---
updated-dependencies:
- dependency-name: org.yaml:snakeyaml
  dependency-type: direct:development
...

Signed-off-by: dependabot[bot] <[email protected]>

* [FLINK-XXXXX] Bump org.yaml:snakeyaml from 2.0 to 2.2

* [FLINK-35038] Fix SnakeYAML usage after version bump

SnakeYAML introduced breaking API changes and behavior changes between 1.31 and
2.2. This commit uses the updated APIs and explicitly allows the global tag
for StreamMetadata (see changed SnakeYAML behavior in [1]).

[1] snakeyaml/snakeyaml@2b8d47c

---------

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Martijn Visser <[email protected]>

Author: 3 people

flink-connector-kafka/pom.xml

@@ -174,7 +174,7 @@ under the License.
         <dependency>
             <groupId>org.yaml</groupId>
             <artifactId>snakeyaml</artifactId>
-            <version>1.31</version>
+            <version>2.2</version>
             <scope>test</scope>
         </dependency>
 

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/YamlFileMetadataService.java

@@ -29,6 +29,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yaml.snakeyaml.DumperOptions;
+import org.yaml.snakeyaml.LoaderOptions;
 import org.yaml.snakeyaml.TypeDescription;
 import org.yaml.snakeyaml.Yaml;
 import org.yaml.snakeyaml.constructor.Constructor;
@@ -252,19 +253,25 @@ Set<KafkaStream> parseFile() throws IOException {
     }
 
     private static Yaml initYamlParser() {
-        Representer representer = new Representer();
+        DumperOptions dumperOptions = new DumperOptions();
+        Representer representer = new Representer(dumperOptions);
         representer.addClassTag(StreamMetadata.class, Tag.MAP);
         TypeDescription typeDescription = new TypeDescription(StreamMetadata.class);
         representer.addTypeDescription(typeDescription);
         representer.setDefaultFlowStyle(DumperOptions.FlowStyle.BLOCK);
-        return new Yaml(new ListConstructor<>(StreamMetadata.class), representer);
+        LoaderOptions loaderOptions = new LoaderOptions();
+        // Allow global tag for StreamMetadata
+        loaderOptions.setTagInspector(
+                tag -> tag.getClassName().equals(StreamMetadata.class.getName()));
+        return new Yaml(new ListConstructor<>(StreamMetadata.class, loaderOptions), representer);
     }
 
     /** A custom constructor is required to read yaml lists at the root. */
     private static class ListConstructor<T> extends Constructor {
         private final Class<T> clazz;
 
-        public ListConstructor(final Class<T> clazz) {
+        public ListConstructor(final Class<T> clazz, final LoaderOptions loaderOptions) {
+            super(loaderOptions);
             this.clazz = clazz;
         }