如何使用 Apache Apex 对 Kafka 0.9 运算符进行单元测试

Question

从 [e-mail] 转发

我想使用支持 0.9 版本协议的新 Kafka 运算符运行单元测试代码。

在这个过程中，我包含了Malhar-Kafka库版本（3.3.1-孵化），并使用Apex引擎（版本3.3.0）作为测试/提供。

编译工作正常，但我的单元测试无法正常运行，并显示" java.lang.ClassNotFoundException： com.datatorrent.lib.util.KryoCloneUtils"异常。

运行使用与 Apex 引擎集成的 Kafka 0.9 运算符的单元测试的推荐方法是什么？我假设Malhar-contrib库Kafka运算符不符合0.9.9。

单元测试代码是这样的：

类CassandraEventDetailsStreamingApp在下面的代码片段中扩展了AbstractKafkaInputOperator。

异常出现在方法 lma.getController（） 中;

@Test
public void testApplication() throws IOException, Exception {
    try {
        LocalMode lma = LocalMode.newInstance();
        Configuration conf = new Configuration(false);
        conf.addResource(this.getClass().getResourceAsStream("/dag-test-props.xml"));
        lma.prepareDAG(new CassandraEventDetailsStreamingApp(), conf);
        LocalMode.Controller lc = lma.getController();
        lc.run();
    } catch (ConstraintViolationException e) {
        Assert.fail("constraint violations: " + e.getConstraintViolations());
    }
}

Answer 1

我能够通过从 Apex-engine 的依赖项部分排除 Malhar-library 和 Malhar-contrib 的依赖项来解决问题，apex-api 。

这使得 Malhar

的 3.3.1 孵化版本进入了类路径，随后是 Malhar-Kafka 库的 3.3.1 孵化版本）。