java - Nifi 自定义 Kafka 处理器代码在有限的时间内工作
问题描述
这是我的自定义 kafka 处理器的代码,它只是从 kafka 主题中消费并产生一些数据
ConsumerRecords<byte[],byte[]> records = consumer.poll(1000);
records.forEach(record -> {
FlowFile flowFile = session.create();
if (flowFile == null) {
return;
}
try {
byte[] outputBytes = (record == null) ? EMPTY_JSON_OBJECT :
genericData.toString(record.value()).getBytes(StandardCharsets.UTF_8);
flowFile = session.write(flowFile, rawOut -> {
rawOut.write(outputBytes);
consumer.commitSync();
});
} catch (ProcessException pe) {
getLogger().error("Failed to deserialize {}", new Object[]{flowFile, pe});
session.transfer(flowFile, REL_FAILURE);
return;
}
flowFile = session.putAttribute(flowFile, "topic", record.topic());
flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "application/json");
getLogger().info("flowFile id " + flowFile.getId());
session.transfer(flowFile, REL_SUCCESS);
});
这段代码需要一批大约 500 条 kakfa 消息,并生成一些 flowFile 用于输出。我需要的显然是将它放在一个while循环中,一遍又一遍地做同样的事情。但是,当我这样做时,处理器中没有任何东西。虽然仍然,信息日志显示 flowFile ids 增加了,并且似乎产生了实际的 flowFile。我测试的一件事是这只发生在无限的while循环中。当我使用有限的 for 循环时,处理器工作正常。我想知道可能有一些关于 nifi flow internal 我不知道的东西。
解决方案
问题是我没有手动提交会话。所以它只有在方法返回时才被提交,这在无限循环的情况下从未发生过。人为的解决方案最终变成了这样。
while(true)
ConsumerRecords<byte[],byte[]> records = consumer.poll(Duration.ofMillis(1000));
records.forEach(record -> {
FlowFile flowFile = session.create();
if (flowFile == null) {
return;
}
try {
byte[] outputBytes = (record == null) ? EMPTY_JSON_OBJECT :
genericData.toString(record.value()).getBytes(StandardCharsets.UTF_8);
flowFile = session.write(flowFile, rawOut -> {
rawOut.write(outputBytes);
consumer.commitSync();
});
} catch (ProcessException pe) {
getLogger().error("Failed to deserialize {}", new Object[]{flowFile, pe});
session.transfer(flowFile, REL_FAILURE);
return;
}
flowFile = session.putAttribute(flowFile, "topic", record.topic());
flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "application/json");
getLogger().info("flowFile id " + flowFile.getId());
session.transfer(flowFile, REL_SUCCESS);
session.commit();
});
}
推荐阅读
- java - Java Junit在预期和实际时失败是一样的吗?
- erlang - 如果重新绑定不会改变变量的现有内存位置,我如何在重新绑定后获取旧值。例如:
- php - 在 Mac OS X 上使用 PHP 的 Heroku Local
- android-studio - Gradle 同步失败:构建文件 '...\build.gradle' 不应包含包语句
- node.js - Google Cloud CDN 无法从 GCS 后端服务访问带有空格的文件
- android - 使用通配符 ssl 证书发出握手错误
- javascript - 我如何在javascript中获取克隆输入值
- r - 字符串参数返回列平均值的函数
- java - void android.widget.Button.setOnClickListener(android.view.View$OnClickListener)' 在一个空对象引用上
- php - 在 PHP 中将多维 json/array 插入多维 json/array