apache-kafka - Alpakka Kafka 无法定义批处理
问题描述
// #atLeastOnceBatch
Consumer.Control control =
Consumer.committableSource(consumerSettings, Subscriptions.topics("topic1"))
.mapAsync(1, msg ->
business(msg.record().key(), msg.record().value())
.thenApply(done -> msg.committableOffset())
)
.batch(
20,
ConsumerMessage::createCommittableOffsetBatch,
ConsumerMessage.CommittableOffsetBatch::updated
)
.mapAsync(3, c -> c.commitJavadsl())
.to(Sink.ignore())
.run(materializer);
// #atLeastOnceBatch
我正在尝试测试驱动 Alpakka Kafka 连接器至少一次批处理示例,我收到以下编译时错误
ConsumerMessage 类型未定义此处适用的 createCommittableOffsetBatch(ConsumerMessage.CommittableOffset)
并且 ConsumerMessage.CommittableOffsetBatch 类型未定义适用于此处的 updated(S, ConsumerMessage.CommittableOffset)
解决方案
这些在 v 0.22 中可用。不幸的是,与 Akka Docs 相比,Alpakka 的文档缺乏一点。
推荐阅读
- javascript - 使用来自外部 PHP 发布请求的数据更新 HTML 内容
- typescript - Webpack 图片找不到模块
- r - 使用 emmeans 的标准化效应量解释
- html - 如何使用 html 和 css 创建一个简单的订阅表单,放在页脚上方?
- google-chrome - 清单:无法使用“chrome_settings_overrides”覆盖 Chrome 主页
- javascript - 我们如何将参数传递给已经构建的 Vue JS 应用程序?
- c++ - 如何在不从文件中读取图像的情况下显示 Mat 图像
- c# - 如何在asp.net c#中调用Javascript函数并传递参数?
- r - 使用 StepAIC 后从模型中提取公式
- ios - 无法在 Swift 中更改 backBarButtonItem