apache-kafka

0热度

1回答

REST Microservice如何与另一个混合的Microservice进行通信，这意味着他可以与REST和Message Queue进行通信。例如一个API网关。对于外部世界，他可以通过REST与应用程序，手机进行通信，但是来自后端的通信是通过消息队列进行的。使用案例：我的主页想要从数据库中获得车辆。他通过GET-Request请求API网关。 API网关接受GET请求并将其发布到消息队

0热度

1回答

Apache Kafka：生产者-Consume API没有在GCP上运行

我一直试图在Kafka集群上运行我的生产者和消费者api，但它不工作。 1）动物园管理员是在一个GCP VM实例运行 2）卡夫卡正在另一个GCP VM实例运行步骤如下：步骤1）使用运行动物园管理员以下命令： bin/zookeeper-server-start.sh config/zookeeper.properties 个Zookeeper.properties： #the direct

0热度

1回答

设置不同的线程用于内卡夫卡不同操作行旅

我已经定义以下在卡夫卡拓扑流 Operation 1 : input_stream ----> filter ----> window_processing ----> write_to_topic Operation 2 : input_stream ----> write_to_topic 我观察到，这两个操作正在由同一线程（即使我增加线程StreamsConfig.NUM_STREAM

0热度

1回答

“为”卡夫卡KStreams循环支持

我需要知道如何使用“为”我的卡夫卡KStreams线环......下面是我的“for”循环需要被列入KStreams for (int i = 0; i < 6 ; i++) { try { textlines.flatMapValues(value -> Arrays.asList(value.split("\\},\\{"))); Thread.s

0热度

3回答

Apache Kafka + Apache Camel集成+ POC问题java.util.Hashtable.put上的java.lang.NullPointerException（Hashtable.java:459）

这是我的路由生成器。在这里，我试图从我的文件插入数据topic.Later，我通过我的主要方法，并使用骆驼上下文我运行它。我尝试了几个代码，但没有人帮助我。我正在开发Apache kafka的POC - 骆驼。 public class SimpleRouteBuilder extends RouteBuilder { @Override public void config

0热度

1回答

Spark Streaming检查点失败后读取

我试图实现包含容错的Kafka应用程序的Spark Streaming。当我重新启动应用程序时，它会读取重新启动前已读取的消息，并且我的计算出错了。请帮我解决这个问题。这是用Java编写的代码。 public static JavaStreamingContext createContextFunc() { SummaryOfTransactionsWithCheckpoints a

0热度

2回答

卡夫卡延迟消息消耗

如何使用Apache Kafka产生/消费延迟消息？看起来像标准的Kafka（和Java kafka-client）功能没有这个功能。我知道我可以用标准的等待/通知机制自己实现它，但它看起来不太可靠，所以我们赞赏任何建议和良好做法。找到related question，但它没有帮助。正如我所看到的：Kafka基于文件系统的连续读取，并且只能用于直接读取保留消息排序的主题。我对吗？

0热度

1回答

连接到安装在HDInsight上的Kafka（Azure）

我需要从外部Java应用程序连接到作为Azure上HDinsight一部分启动的Kafka集群。我有3个经纪人实例，3个ZooKeeper和一个ZooKeeper客户端。现在我的问题：如何指定代理连接字符串。在管理面板上，我可以看到3个经纪人：xxx-1.yyy.zzz.internal.cloudapp.net,xxx-2.yyy.zzz.internal.cloudapp.net - 但这些

0热度

1回答

春云流Kafka粘结剂：“尝试从状态IN_TRANSACTION到状态IN_TRANSACTION的无效转换”

我正在尝试使用Spring云流+ Kafka绑定对Apache Kafka进行“恰好一个交付”概念的一些PoC。我安装了Apache Kafka“kafka_2.11-1.0.0”，并在生产者中定义了“transactionIdPrefix”，我知道这是我在Spring Kafka中启用事务所需要做的唯一事情，但是当我这样做时，运行简单源代码&在同一个应用程序中接收器绑定，我看到一些消息在消费者

0热度

1回答

Kafka Streams：使用相同的`application.id`来消费多个主题

我有一个应用程序需要监听多个不同的主题;每个主题都有独立的消息处理逻辑。我曾经想过为每个KafkaStreams实例使用相同的kafka属性，但是我得到如下所示的错误。错误 java.lang.IllegalArgumentException: Assigned partition my-topic-1 for non-subscribed topic regex pattern; subscr