[FLINK-16262][connectors/kafka] Set the context classloader for paral…

…lel stream in FlinkKafkaProducer

This closes apache#11247

flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java

@@ -49,6 +49,7 @@
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.IOUtils;
 import org.apache.flink.util.NetUtils;
+import org.apache.flink.util.TemporaryClassLoaderContext;
 
 import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
 
@@ -1095,18 +1096,24 @@ private void resetAvailableTransactionalIdsPool(Collection<String> transactional
  // ----------------------------------- Utilities --------------------------
 
  private void abortTransactions(final Set<String> transactionalIds) {
+ final ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
  transactionalIds.parallelStream().forEach(transactionalId -> {
- // don't mess with the original configuration or any other properties of the
- // original object
- // -> create an internal kafka producer on our own and do not rely on
- // initTransactionalProducer().
- final Properties myConfig = new Properties();
- myConfig.putAll(producerConfig);
- initTransactionalProducerConfig(myConfig, transactionalId);
- try (FlinkKafkaInternalProducer<byte[], byte[]> kafkaProducer =
- new FlinkKafkaInternalProducer<>(myConfig)) {
- // it suffices to call initTransactions - this will abort any lingering transactions
- kafkaProducer.initTransactions();
+ // The parallelStream executes the consumer in a separated thread pool.
+ // Because the consumer(e.g. Kafka) uses the context classloader to construct some class
+ // we should set the correct classloader for it.
+ try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(classLoader)) {
+ // don't mess with the original configuration or any other properties of the
+ // original object
+ // -> create an internal kafka producer on our own and do not rely on
+ // initTransactionalProducer().
+ final Properties myConfig = new Properties();
+ myConfig.putAll(producerConfig);
+ initTransactionalProducerConfig(myConfig, transactionalId);
+ try (FlinkKafkaInternalProducer<byte[], byte[]> kafkaProducer =
+ new FlinkKafkaInternalProducer<>(myConfig)) {
+ // it suffices to call initTransactions - this will abort any lingering transactions
+ kafkaProducer.initTransactions();
+ }
  }
  });
  }