如何解决KafkaStreams Scala:替换 FunctionsCompatConversions
我在现有的一堆应用程序中使用 KafkaStreams 2.6.0、scala。
我正在设计一个方案来最大化正常运行时间/稳健性,并且在每个异常时,记录 +(丢弃或发送到死信主题)。我想在不在整个应用程序中显式添加 Try/try-catch 块的情况下执行此操作。
我有这个想法,用我自己的 FunctionsCompactConversions
替换 GuardedFunctionsCompactConversions
,并在那里添加 Try-s
例如
替换
implicit class ForeachActionFromFunction[K,V](val p: (K,V) => Unit) extends AnyVal {
def asForeachAction: ForeachAction[K,V] = (key: K,value: V) => p(key,value)
}
与
implicit class ForeachActionFromFunction[K,value: V) => {
setLogContext(value,key)
Try(p(key,value)) match {
case Success(_) =>
case Failure(ex) => Error(s"asForeachAction Failed $ex when handle ($key,$value)")
}
}
}
或
def asPredicate: Predicate[K,value)
与
def asPredicate: Predicate[K,value)) match {
case Success(s) => s
case Failure(ex) =>
Error(s"asPredicate Failed $ex when handle ($key,$value)")
false
}
}
等等。这样 -
-
应用程序提供的所有代码都受到保护(谓词、reducers、Serde 等),并且不能“忘记”尝试/捕获
-
集中能够在生产等中禁用此日志记录
-
如果进行故障排除,可以集中选择 Guarded 版本,而默认情况下使用紧凑版本
不幸的是,我没有找到正确的方法来做到这一点。我创建了 GuardedFunctionsCompactConversions
对象,但无法扩展/覆盖紧凑的对象,也无法将其导入正确的 KTable/KStream/KGroupedStream/... 类。
这是一个常见的要求吗?我希望它是。 有没有正确的方法可以到达那里?
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。