[FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer

[hugogu]

So that when connector-kafka installed into Flink Libs, it can work with a Customer Deserializer class defines in User Code.

This is useful with work with Spring Kafka JsonDeserializer. JsonDeserializer is a generic class and can't be used to build a KafkaSource directly because the Generic Type information required would lost.

A typical use case of JsonDeserializer looks like following

• Declare a sub class of JsonDeserializer with customized object mapper (which is in User Jar for sure)

  class AccountBalanceActivityDeserializer :
      JsonDeserializer<AccountBalanceActivity>(SpringObjectMapperSupplier.INSTANCE)

• Use it to build a KafkaSource

      @Bean
      fun source(): KafkaSource<AccountBalanceActivity> {
          return KafkaSource.builder<AccountBalanceActivity>()
              .setBootstrapServers("kafka:9092")
              .setTopics("BalanceActivity")
              .setGroupId("FraudDetection")
              .setStartingOffsets(OffsetsInitializer.latest())
              .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(AccountBalanceActivityDeserializer::class.java))
              .build()
      }

This fix is to make above sample use pattern work when connector-kafka installed into Flink Libs rather than included in user jar.

[boring-cyborg]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[MartijnVisser]

Can we somehow add a test for this? I'm not too fond of single line changes without actually adding a test

[hugogu]

@MartijnVisser Thanks for the quick response. There are existing tests covering this class and exactly this line of code and I also verified it pass. I'm happy to add more tests to ensure the user code class loader is being used.

[MartijnVisser]

There are existing tests covering this class and exactly this line of code and I also verified it pass. I'm happy to add more tests to ensure the user code class loader is being used.

I don't think that's correct, because if there were existing tests I would have expected those to fail because of your change.

[hugogu]

@MartijnVisser True. I added a couple of tests that would fail without my change. Hope it looks good now. Please let me know if anything else is needed.

[hugogu]

@MartijnVisser May I have your advice on the use of TemporaryClassLoaderContext in the implementation? It looks unnecessary to me. Do you think I should just remove it in this PR?

    public void open(InitializationContext context) throws Exception {
        final ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
        try (TemporaryClassLoaderContext ignored =
                TemporaryClassLoaderContext.of(userCodeClassLoader)) {
            serializer =
                    InstantiationUtil.instantiate(
                            serializerClass.getName(),
                            Serializer.class,
                            userCodeClassLoader);

            if (serializer instanceof Configurable) {
                ((Configurable) serializer).configure(config);
            } else {
                serializer.configure(config, isKey);
            }
        } catch (Exception e) {
            throw new IOException("Failed to instantiate the serializer of class " + serializer, e);
        }
    }

[hugogu]

@MartijnVisser I also fix a few style errors, please let me know if it is good now or any other change to make. Thanks for your review.

[morazow]

Thanks @hugogu for the PR! I have added some comments, please have a look.

Do you know why the CI is failing?

[morazow]

Looks good, thanks @hugogu for quick update! Hopefully the CI is green this time 🤞

[hugogu]

@morazow Thanks for your approval. I have just rebased this PR to include the build fix made in main branch. Hopefully the build would success this time.

I also noticed that v3.1 branch, may I know if this fix can be included? Shall I raise another PR for that? Not quite sure about the release procedure yet.

cc @MartijnVisser

[hugogu]

Hi @uce and @rmetzger, do you mind reviewing the change and let me know if any change is required before merging. Thanks.

[hugogu]

Is there any other suggestions or problems I can fix before getting this PR merged please?