Kafka用例：连续读取Kafka，对消息进行解密，然后插入db
Kafka用例：连续读取Kafka（可能是一个流），用Java方法对消息值执行解密，然后插入到db（可以是接收器连接器，也可以是我有一个Java REST API，如果需要的话可以插入到db）
对于这个用例，我还没有看到任何可行的解决方案，因为：
>
我不能用Kafka存储解密数据。
我需要在插入到数据库之前执行解密步骤。
看看Kafka Connect和Kafka Streams来解决用例：
我似乎不能使用Kafka Connect，因为我不知道如何包含解密步骤。
看来我不能用Kafka流，因为1。它是为阅读和写回主题而构建的。2.即使我使用处理器API实现了自定义逻辑，我也看不出如何将KStream消息值转换为String来传递给我的Java解密方法或数据库。
我有一个工作的解决方案，在Java应用程序中使用常规的Kafka消费者，但它是作为一个1 off批处理作业，我需要一个应用程序，是长期生存的，并不断检查Kafka队列。我可以无限地循环批处理作业，但我不确定这是一个可行的选项。
共有1个答案
您应该使用Kafka Connect并通过实现转换接口编写自定义SMT：
https://github.com/apache/kafka/blob/trunk/connect/api/src/main/java/org/apache/kafka/connect/transforms/transform.java
更多详细信息请参阅本讲座：https://www.confluent.io/thank-you/single-message-transformations-not-transformations-youre-look/
-
我试图解析一条Kafka消息，它是以某种加密的AVRO格式。我有以下AvroSchema。avsc avro架构文件： 现在，我编写了以下代码来获取JSON格式的数据： 请帮我解密这封信。 加密字节消息属于以下类型：<代码>080-21-0001：�哦�@@��A.�ǐ�U：�哦�@@��A 我按照建议进行了更改，现在我有以下代码： 但我仍然得到错误为“不是数据文件”。
-
我刚开始学Kafka，Kafka-蟒蛇。在下面的代码中，我试图在消息到达时读取它们。但出于某种原因，消费者似乎要等到一定数量的消息积累后才能获取它们。 我最初以为是因为正在批量出版的制片人。当我运行“kafka-console-consumer--bootstrap-servers--topic”时，我可以看到发布后收到的每一条消息（就像在consumer控制台上看到的那样） 有人能指出用KafK
-
我有一个Kafka集群（版本：0.10.1.0），有9个代理和10个分区。 我尝试使用camel kafka从java应用程序中获取消息。这是我的pom。xml 这只是我使用的与骆驼Kafka相关的依赖项。下面是骆驼Kafka消费者代码。 我正在使用文档中指定的KafkaURIhttps://camel.apache.org/components/latest/kafka-component.ht
-
我目前正在使用Kafka listener设置Spring Boot应用程序。我试图只对消费者进行编码。对于producer，我现在正在从Kafka控制台手动发送消息。我举了一个例子：http://www.source4code.info/2016/09/spring-kafka-consumer-producer-example.html 我尝试将其作为Spring Boot应用程序运行，但无法
-
有没有解决这个问题的方法？？？我无法读取KAFKA-AVRO架构消息。我正在尝试将消息从logstash发送到KAFKA到hdfs。 以下是技术堆栈： LogStash 2.3-当前生产版本 汇流3.0。 插件：A。Logstash-kafka-Output插件B。logstash-codec-avro。 动物园管理员:3.4.6 Kafka:0.10.0.0 Logstash配置文件如下所示：
-
我有一个Java Spring Kafka应用程序，它将Book类型的对象发送到Kafka主题。然后我试着用Kafka流来映射每一条信息，把这本书的作者作为它的关键字。然后，我尝试将它们添加到一个KTable中，该表保存了密钥和拥有该密钥的消息的数量。然后，该表被发送到一个输出kafka主题。 书籍型号： 流结构： 运行应用后，输出显示每个唯一键，但随机表情，而不是计数。 在搞砸之后，我发现了一件