卡夫卡连接3.3中的ClassLoading

Question

我们已经修改了Kafka，以支持自定义转换器，该转换器将将单个SinkRecord转换为多个SinkRecords，从而支持事务性插入。在创建接收器时，可以在配置属性中指定实现SinkRecordConverter的类。

然后，我们尝试用这个自定义转换器的实现打包一个uber jar，并尝试以两种方式部署它：

1. 我们放置在kafka-connect jdbc的同一个文件夹中。
2. 我们将connect-distributed.properties中的plugins.path修改为/usr/local/share/java，并将转换器放在plugins.path中

然后，我们尝试部署接收器，但在这两种情况下，尝试通过反射创建此转换器实例的代码都在java.lang.ClassNotFoundException.中失败。

我们试图调试类加载问题，方法是放置一个断点，在这两种情况下都会出现问题：

• 在第一种情况下，jar将显示为URLClasspath上的一个jar。
• 在第二种情况下，它甚至不会作为URLClasspath上的一个jars出现。

向kafka-connect jdbc添加自定义转换器的正确方法是什么？

Answer 1

我们有两个问题在一个：

1. 为了组装jars，我们使用了一个名为oneJar的SBT插件，它创建了一个定制的类加载器。
2. 我们需要从现有的kafka连接器(jdbc)中访问该类，而不仅仅是从kafka连接。

我们找到的解决方案如下：

• 我们放弃了uber jar，并使用sbt在kafka-connect实例上部署了所有的库。
• 我们将罐子物理地放置在kafka-connect jdbc所在的文件夹中。