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
      }
    }

等等。这样 -

1. 应用程序提供的所有代码都受到保护（谓词、reducers、Serde 等），并且不能“忘记”尝试/捕获

2. 出现任何错误/异常时，可以记录手头的消息，提供有关修复内容的见解

3. 集中能够在生产等中禁用此日志记录

4. 如果进行故障排除，可以集中选择 Guarded 版本，而默认情况下使用紧凑版本

不幸的是，我没有找到正确的方法来做到这一点。我创建了 GuardedFunctionsCompactConversions 对象，但无法扩展/覆盖紧凑的对象，也无法将其导入正确的 KTable/KStream/KGroupedStream/... 类。

这是一个常见的要求吗？我希望它是。 有没有正确的方法可以到达那里？

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）