从 users@apex.incubator.apache.org 转发
我想使用支持 0.9 版本协议的新 Kafka Operator 运行单元测试代码。
在这个过程中,我包含了 Malhar-Kafka 库版本( 3.3.1-incubating )并使用 Apex 引擎(版本 3.3.0 )作为测试/提供。
编译工作正常,但我的单元测试无法正常运行,出现“java.lang.ClassNotFoundException: com.datatorrent.lib.util.KryoCloneUtils”异常。
运行使用与 Apex 引擎集成的 Kafka 0.9 运算符的单元测试的推荐方法是什么?我假设 Malhar-contrib 库 Kafka 运算符不兼容 0.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());
}
}