我们在kafka中使用ktabke进行聚合，它是kafka文档中非常基本的用法。
我只是想调查一下，如果在聚合时某些消息消耗失败，我们如何将这些消息移动到错误主题或dlq。
我在kstream中找到了类似的东西，但在ktable中找不到，而且我无法简单地将kstream解决方案扩展到ktable。
使用kafka的streams api处理错误消息的kstream参考
我的用例对于任何类型的异常都非常简单，只需转到错误主题并转到不同的消息
我们在kafka中使用ktabke进行聚合，它是kafka文档中非常基本的用法。
我只是想调查一下，如果在聚合时某些消息消耗失败，我们如何将这些消息移动到错误主题或dlq。
我在kstream中找到了类似的东西，但在ktable中找不到，而且我无法简单地将kstream解决方案扩展到ktable。
使用kafka的streams api处理错误消息的kstream参考
我的用例对于任何类型的异常都非常简单，只需转到错误主题并转到不同的消息
1条答案
按热度按时间cld4siwp1#
atm（kafka2.2）没有内置的对您所请求的内容的支持，但是您需要确保您的应用程序代码不会抛出任何异常。提供的所有可以配置的处理程序都是针对kafka streams运行时抛出的异常的。这些处理程序正在提供，因为否则用户根本没有机会对这些异常做出React。
请随意创建特性请求jira。