我在现有的一系列应用程序中使用kafkastreams2.6.0和scala。
我正在设计一个方案，以最大限度地提高正常运行时间/健壮性，并对每个异常使用log+（丢弃或发送到死信主题）。我不想在应用程序中显式地添加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, V](val p: (K, V) => Unit) extends AnyVal {
    def asForeachAction: ForeachAction[K, V] = (key: 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, V] = (key: K, value: V) => p(key, value)

具有

def asPredicate: Predicate[K, V] = (key: K, value: V) => {
      setLogContext(value, key)
      Try(p(key, value)) match {
        case Success(s) => s
        case Failure(ex) => 
          Error(s"asPredicate Failed $ex when handle ($key, $value)")
          false
      }
    }

等等，这边-
所有应用程序提供的代码都是受保护的（ predicate 、约简、serde等等），不能“忘记”try/catch
在出现任何错误/异常时，您可以记录手头的消息，从而了解应该进行哪些修复
能够在生产等中集中禁用此日志记录。
如果进行故障排除，可以集中选择保护版本，而默认情况下使用紧凑版本
不幸的是，我没有找到正确的方法来做这件事。我创造了 GuardedFunctionsCompactConversions 对象，但无法扩展/重写压缩对象，也无法将其导入正确的ktable/kstream/kgroupedstream/。。。班级。
这是一个共同的要求吗？我想是的。有没有正确的方法去那里？