KafkaStreams在拓扑中添加多个处理器不起作用
KafkaStreams是一款开源的流处理平台，用于在Apache Kafka消息队列上执行实时流处理任务。它可以以低延迟、高吞吐量和可伸缩性的方式处理数据流，并支持对流数据进行转换、过滤、聚合等操作。KafkaStreams使用Java编程语言进行开发，提供了丰富的API和功能，方便开发人员构建复杂的流处理应用。
在KafkaStreams中，拓扑（Topology）是指一个数据流处理应用的逻辑结构。拓扑由多个处理器（Processor）组成，每个处理器负责执行特定的流处理操作。处理器可以通过连接器（Connector）将其连接起来，形成数据流的处理流程。然而，如果在拓扑中添加多个处理器后发现不起作用，可能是由于以下几个原因：
- 拓扑结构问题：首先，需要确保处理器的连接顺序正确。每个处理器的输出必须与下一个处理器的输入相连接，形成连续的数据流。如果连接顺序不正确，数据流可能无法顺利地传递到下一个处理器，导致不起作用。可以通过仔细检查处理器之间的连接关系来解决该问题。
- 处理器配置问题：其次，需要检查每个处理器的配置是否正确。每个处理器都有自己的配置参数，例如输入主题、输出主题、窗口大小、处理逻辑等。如果配置参数不正确或不一致，处理器可能无法正确地处理数据流。可以通过查看每个处理器的配置参数，并确保其与实际需求相匹配来解决该问题。
- 数据流问题：最后，需要检查数据流是否按预期到达每个处理器。KafkaStreams基于Kafka消息队列进行流处理，因此可能存在消息传递延迟或顺序不一致的情况。可以通过检查输入主题的消息产生情况、消费情况以及消息顺序来判断数据流是否正常。
为了更好地使用KafkaStreams，腾讯云提供了云原生的消息队列服务——CMQ（Cloud Message Queue），用于在云端和分布式系统之间进行可靠的消息传递。CMQ提供了高性能、低延迟的消息队列服务，并与KafkaStreams相互兼容。腾讯云CMQ产品可以帮助用户构建可靠的流处理应用，实现实时数据处理和分析。
相关产品介绍链接地址：腾讯云CMQ产品介绍
相关·内容
我有一个Spring boot应用程序，它使用处理器API来生成拓扑和指向相同拓扑的addGlobalStateStore。
我想创建另一个拓扑(因此也是另一个KafkaStreams)，以便从另一组主题中读取内容，并希望在新拓扑中共享以前创建的存储。共享的意思是，两种拓扑的底层状态存储应该是相同的。从一个拓扑写入的任何数据在另一个拓扑中都应可见。
如果不编写包装器端点来访问状态存储(例如REST调用)，这是可能的吗？
或者我的用例需要一个外部状态存储，比如redis？
浏览 6提问于2019-03-14得票数 1
我有一个拓扑：
Topology builder = new Topology();
builder.addSource("source",stringDeserializer,stringDeserializer, "TOPIC-DEV-ACH")
.addProcessor("process1", ProcessorOne::new , "source")
.addProcessor("process2", ProcessorTwo::new ,
浏览 11提问于2019-11-08得票数 1
1回答
我一直在尝试Apache Kafka开源生产者和消费者数据流。我有一个问题，使用1.) KafkaProducer, producer.send和2.) Streamsbuilder, KafkaStreams代码有什么区别？它们不都是用于kafka流数据的吗？如果我只使用KafkaProducer, producer.send，我不能实现数据流吗？
浏览 1提问于2018-03-16得票数 0
2回答
我有下面的kafka流代码
public class KafkaStreamHandler implements Processor<String, String>{
private ProcessorContext context;
@Override
public void init(ProcessorContext context) {
// TODO Auto-generated method stub
this.context = context;
}
public KeyV
浏览 3提问于2020-04-17得票数 7
回答已采纳
所以我的问题是，我在我的属性文件中定义了一些Kafka主题，并且我可以在我的SpringBoot应用程序中从该主题读取KafkaStream<String, String>。但是我想访问KafkaStreams对象，这样我就可以打印我的KafkaStreams拓扑，这对开发很有用。在我的一个@StreamListener中，我尝试检索流生成器进程bean，这样我就可以以这种方式获取底层的KafkaStreams对象(如这里所描述的：https://cloud.spring.io/spring-cloud-stream-binder-kafka/spring-cloud-strea
浏览 46提问于2020-07-08得票数 0
我在本地机器上运行多个kafka流消费者实例(2个实例)，每个实例都有自己的自定义本地存储，并且每个实例都有不同的名称。
根据文档，如果其中一个实例宕机，那么kafka必须将死实例的存储同步到活实例的存储(如果我错了，请纠正我)。
我已经为两个实例配置了相同的应用id，让kafka知道这些实例属于同一个组。
当其中一个实例被杀死时，其他(活的)实例的存储不会与死实例的存储同步。我已经在两个商店上启用了更改日志主题。
但是，当我在两个实例中具有相同的商店名称时，商店会按预期同步，不确定这些实例是否指向同一个商店。对于这两个实例，我有不同的StreamsConfig.STATE_DIR_CONFI
浏览 1提问于2019-06-21得票数 0
很抱歉，如果之前已经讨论过这件事，我找不到任何与此密切相关的东西。我有这个Kafka应用程序，它从多个主题中读取，在DB上保存记录，然后将事件发布到输出主题。很简单，就卡夫卡当地商店而言，它是无状态的。(以下拓扑结构)
Topic1(T1)有5个分区，Topic2(T2)有一个分区。这里的问题是，在使用两个主题的同时，如果我想“全速”使用T1 (5个消费者)，这并不能保证我会为T1上的每个分区拥有专门的消费者。它将分布在两个主题分区中，我可能最终会遇到不平衡的使用者(和空闲的消费者)，如下所示：
c1: t1p1，t1p3，c2: t1p2，t1p5，c3: t1p4，t2p1，c4：(空闲消
浏览 2提问于2020-06-01得票数 0
回答已采纳
最近，我一直在学习如何使用Kafka客户端，而我一直在努力解决的一件事是，如何使用Redis这样的工具从默认状态存储(RocksDB)切换到自定义状态存储。汇合文档清楚地表明，您必须为您的自定义存储实现StateStore接口，并且您必须为创建该存储的实例提供StoreBuilder的实现。
这是我到目前为止为我的定制商店准备的东西。我还添加了一个简单的写方法，通过Redis命令将一个新条目附加到指定的流中。
public class MyCustomStore<K,V> implements StateStore, MyWriteableCustomStore<K,V>
浏览 3提问于2020-08-07得票数 0
回答已采纳
这是关于升级生产中的现有代码库，这些代码库使用来自kafka-clients，kafka-streams，spring-kafka 2.4.0到2.6.x的窗口，并将spring-boot-starter-parent2.2.2.RELEASE升级到2.3.x，因为2.2与kafka-streams 2.6不兼容。
现有的代码在下面的旧版本中提到了这些bean(2.4.0，2.2春季版本)：
@Bean("DataCompressionCustomTopology")
public Topology customTopology(@Qualifier("CustomFa
浏览 20提问于2021-05-15得票数 2
1回答
我对卡夫卡和卡夫卡流很陌生，所以请容忍我。我想知道我在这里是否走对了路。
我现在写的是一个Kafka主题，并试图通过rest服务访问数据。在访问原始数据之前，需要对其进行转换。
到目前为止，我拥有的是一个将原始数据写入主题的生产者。
1.)现在我希望streams应用程序(应该是一个在容器中运行的jar )只转换我想要的形状的数据。这里遵循物化的视图范例。
过简化版1.)
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source =
builder.stre
浏览 3提问于2017-08-11得票数 1
如何在应用程序运行时添加传入主题和更改传出主题？根据当前正在处理的传入主题，传出主题应该发生变化。 in_topic1 -> filter OK -> out_topic1；in_topic2 -> filter OK -> out_topic2。 final Serde<byte[]> byteArraySerde = Serdes.ByteArray();
final Serde<String> stringSerde = Serdes.String();
final StreamsBuilder builder = new
浏览 10提问于2020-12-12得票数 0
最近我正在学习Apache Kafka流，播放世界计数examples.Below是我的代码。
public class StreamsStarterApp {
public static void main(String[] args) {
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-starter-app");
properties.put(StreamsConfig.BOOTSTRAP_S
浏览 2提问于2020-08-16得票数 2
回答已采纳
1回答
我对从Stream内部调用交互式查询有一个特殊的要求。这是因为我需要创建一个新的Stream，它应该在State Store中包含数据。截断代码如下：
tempModifiedDataStream.to(topic.getTransformedTopic(), Produced.with(Serdes.String(), Serdes.String()));
GlobalKTable<String, String> myMetricsTable = builder.globalTable(
topic.getTransformedTopic(),
Material
浏览 0提问于2019-02-21得票数 2
1回答
我想在Kafka Streams的列表中部分收集项目，在给定的时间段或在1000个项目被处理后。
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> kStream = builder.stream(topicTwo);
KTable<String, String> kTable = builder.table(topicOne);
kStream.join(kTable,
(streamValue,
浏览 0提问于2018-04-26得票数 2
因此，我们启动了一堆Kafka流应用程序，而没有意识到默认的复制因子是1。
我们已经做了代码修改(例如 )
但是，我认为这无助于已经部署的应用程序，也无助于更改已经创建的内部主题。
例如，我使用kafkacat列出了一些主题(基于application.id前缀，并且都有一个副本)。
显然，当代理开始出现问题(此处为broker.id 11或21 )时，应用程序无法正常工作。
topic "appid-KTABLE-SUPPRESS-STATE-STORE-0000000013-changelog" with 1 partitions:
partition 0, l
浏览 0提问于2019-08-02得票数 4
1回答
我有一个KStream应用程序，它包含大量的KStreams、joins和其他操作。我使logging.level.org.springframework.kafka.config=debug能够验证正在生成的拓扑，并发现了许多根本没有意义的节点。
然后，我将应用程序简化为以下内容：
interface ShippingKStreamProcessor {
@Input("input")
fun input(): KStream<Int, Customer>
}
@Suppress("UNCHECKED_CAST")
@Conf
浏览 2提问于2019-04-30得票数 1
回答已采纳
我正在学习kafka streams，并写了一个简单的应用程序，代码片段如下：
MainApp:
Topology topology = new Topology();
topology.addSource("SOURCE", "source-topic");
topology.addProcessor("Processor1", () -> new Processor1(), "SOURCE");
topology.addProcessor("Pro
浏览 129提问于2019-06-05得票数 0
对于在Kafka绑定器中生成/使用的消息头的管理，有一个KafkaHeaderMapper接口，它作为bean的实现可以配置以下属性：spring.cloud.stream.kafka.binder.headerMapperBeanName。
Stream中的KafkaStreams绑定器有类似的地方吗？我的意图是能够控制如何在流输入和输出中反序列化/序列化或包括/排除消息头。有人知道如何意识到这一点吗？
浏览 17提问于2022-01-18得票数 0
如果我创建一个简单的拓扑，其中我有一个源和一个处理器，我在控制台中得到的StreamThread是预期的两倍。 例如，如果我将线程设置为1，并且有一个分区，我会看到2个流线程。如果我设置为20个线程，并且有20个分区，我会看到40个流线程。 基于Kafka Streams thread number，我预计这些数字的一半。 我配置了错误的东西，或者这是意料之中的？ 编辑:在stream = new拓扑(KafkaStreams，streamsConfig)；被调用后，我只看到它创建了20个线程。 在调用stream.start()之后，我看到这20个线程从创建状态转换到运行状态。 只有在初始
浏览 25提问于2019-01-03得票数 0
回答已采纳
这是我的转换器：
public class DataEnricher implements Transformer < byte[], EnrichedData, KeyValue < byte[], EnrichedData >> {
private ManagedChannel channel;
private InfoClient infoclient;
private LRUCacheCollector < String,
InfoResponse > cache;
public DataEnricher
浏览 30提问于2020-03-06得票数 0
回答已采纳
我有一个应用程序是基于spring boot，spring-kafka和kafka-streams。当应用程序启动时，它使用默认的主题列表创建kafka streams拓扑。我需要做的是在运行时编辑/重新创建拓扑。例如，当应用程序已经运行时，有一个新的主题名称出现，我想将此主题添加到我的拓扑中。目前我正在考虑以某种方式删除现有的拓扑，关闭并清理KafkaStreams，运行逻辑，我创建拓扑，但使用新的主题名称，并再次启动KafkaStreams。我不想重新启动我的应用程序。有人能建议我如何在运行时做到这一点吗？
浏览 16提问于2020-04-11得票数 1
回答已采纳
我想加入两个卡夫卡主题。一个是KStream，另一个是KTable。左联接抱怨处理器的状态存储不存在。我确实看过许多卡夫卡GitHub和其他地方的代码示例，在这些地方，KStream客户端代码没有显式地创建StateStore。请告知下面的代码中缺少了什么。
应用程序流将与user表连接，以便与app和user一起发出记录。应用程序的所有者是用户。
版本: 1.1.0
谢谢
public void process() {
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID
浏览 0提问于2018-07-28得票数 1
回答已采纳
我在拓扑中添加了多个处理器。只有一个处理器工作正常。我能够接收来自topic1和topic2的消息，并使用Processor1和prod输出到topic3-sink来处理它们。在我的代码中没有调用Processor2。 @Component
public class CustomTopology {
@PostConstruct
public void createTopology() {
Topology topology = new Topology();
topology.addSource("source", new K
浏览 15提问于2019-05-26得票数 2
在/tmp/streams-my-application-id下，我找到了RocksDB使用的文件。我的目的是通过du -h检查文件大小。 当看到文件名时，我对文件名的含义很好奇。文件名的含义是什么？我认为这与Kafka Streams任务和分区有关。 前缀0和1是否表示使用的主题数量，而后者是使用的分区？ 这个KafkaStreams应用程序简单地使用KStream-KTable连接两个主题，其中一个主题是重新分区并还原为KTable。 8,0K ./0_2
8,0K ./0_1
3,1M ./1_2/rocksdb/KSTREAM-REDUCE-STATE-STORE-
浏览 23提问于2019-06-14得票数 1
回答已采纳
我是kafka流处理器的新手，偶然发现了“拓扑”的关键概念。 我创建了一个源代码处理器，它从一个" source -topic“中读取数据，如下所示： Topology topology = new Topology();
topology.addSource("SOURCE", "source-topic"); 上面的代码片段将创建(如果我的理解正确的话)一个名为" source“的源流处理器，并将侦听kafka主题"source- topic”。 我没有为这个“源”流处理器写任何代码，它怎么能从kafka主题中获取消息呢？它是一种由
浏览 32提问于2019-03-19得票数 1
回答已采纳
我对卡夫卡和溪流很陌生。我正在创建一个本地商店，以保存来自特定主题components的所有更新。我不知道我做错了什么。还有其他方法可以从Stream创建存储吗？
我需要在卡夫卡中创建一个主题comp-store吗？
public class MyStream {
final static CountDownLatch latch = new CountDownLatch(1);
private static final String APP_ID = "MyTestApp";
public static void main(String[] arg
浏览 0提问于2019-05-09得票数 0
回答已采纳
1回答
推荐使用哪一种: 1.单个kafka流从多个主题中消费2.不同的kafka流从不同的主题中消费(我已经使用过它，没有遇到任何问题)
是否有可能达到#1？如果是的话，这意味着什么？如果我使用'EXACTLY_ONCE‘设置，它会带来什么样的复杂性？
kafka版本:2.2.0CP2
浏览 1提问于2020-06-18得票数 0
1回答
我试图在应用程序中配置一个全局状态存储，但是它失败了：
线程"svc-toa-app-datainteg-io-dev-fd2e1a47-0758-4b95-9b2e-022ca4df95ca-流-1“org.apache.kafka.streams.errors.TopologyException:无效拓扑中的异常:主题不为拓扑所知。如果同一应用程序的不同KafkaStreams实例执行不同的拓扑，则可能会发生这种情况。请注意，只有当所有运算符按相同的顺序添加时，拓扑才是相同的。
在正式文档之后，我添加了以下内容来配置状态存储：
@Bean
public StreamsBu
浏览 1提问于2022-02-08得票数 0
我正在构建一个拓扑，并希望使用将一些中间值写入数据库。此步骤不改变数据的性质，并且完全无状态。
添加一个需要创建一个，并将这个实例与状态存储的名称一起传递给KStream.process()函数。这就是我不明白的。
如何将对象添加到拓扑中，因为它需要
当应用程序启动时，未能添加上述StateStore将导致此错误：
线程"main“org.apache.kafka.streams.errors.TopologyBuilderException:无效拓扑构建中的异常:尚未添加StateStore。
为什么处理器必须有状态存储？对于无状态且不维护状态的处理器，这似乎很可能是可选的。
浏览 5提问于2016-08-22得票数 6
回答已采纳
注意:这是一个无效的问题。请忽略。
我有一个KSA监听两个主题(两个子拓扑)，一个(子拓扑A)写入状态存储，另一个(子拓扑B)从状态存储读取。
写
...
stream
.mapValues(v -> new Version(v.getHeader().getOccurredAt().getSeconds(), v.getVersion().getValue()))
.groupByKey()
.aggregate(
() -> new Version(0,0),
(aggKey, newValue, aggV
浏览 0提问于2020-02-24得票数 0
我编写了一个Streams Binder应用程序，该应用程序将多个Kafka输入主题复用到一个流中：
spring:
cloud:
stream:
bindings:
process-in-0:
destination: test.topic-a,test.topic-b
(资料来源：)
但是，每当我在输入目标中设置多个主题(用逗号分隔)时，就会发生以下错误：
2022-06-17 14:07:07.648 INFO --- [-StreamThread-1] o.a.k.clients.consumer.KafkaConsumer
浏览 16提问于2022-06-17得票数 0
回答已采纳
在我的Java应用程序中，我有一个Kafka处理器。
我的流程方法如下所示：
@Override
public void process(String key, String value) {
System.out.println("In the process method, the offset is: " + context.offset());
//Some more code
}
其中上下文是init方法中的ProcessorContext。
我启动应用程序并记录如下：
In the process method, the offset is: 120
浏览 0提问于2019-04-22得票数 1
回答已采纳
如何从KafkaStream获取主题名称和分区id。对于任何其他的Kafka用户，我们可以获得主题名称和partitionId，如下所示：
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {System.out.printf("consumed: key = %s, value = %s, partition id= %s, offset = %s%n",recor
浏览 2提问于2020-06-20得票数 4
回答已采纳
使用Kafka，我总是使用以下代码从引用的紧凑主题初始化我的存储：
builder.globalTable(kafkaTopic, Materialized.as("storeMerchant"));
Topology topology = builder.build();
KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();
我想在做商店之前过滤kafkaTopic主题，以消除一些不必要的商家。
就像这样：
GlobalKTable<String, MerchantAvro&
浏览 0提问于2018-06-07得票数 5
回答已采纳
1回答
问题：在一个JVM中拥有一个org.apache.kafka.streams.KafkaStreams类的多个实例意味着什么(EG:内存、CPU使用、并发问题)？
后台：我试图提供一个散列机制，这样如果流操作抛出异常，它就不会将整个KafkaStreams实例转换为ERROR状态。我将应用程序划分为不同的KafkaStreams实例，每个实例负责不同的任务(日志记录、外部web调用、db调用等)。
我还没有找到有关以下方面的文档：(1)从KafkaStreams状态恢复ERROR实例；(2)设计一个应用程序，该应用程序使用KafkaStreams或(3)证明或反驳我的当前方法。
如果我的方法违
浏览 2提问于2018-03-15得票数 1
回答已采纳
在创建处理器API Topology时，我注意到Topology#addStateStore(StoreBuilder, String...)接受多个处理器，这意味着一个状态存储可以由多个处理器共享。
这个设计有什么要注意的地方吗？如果密钥不存在，而实际上其他处理器只是存储该键的值，那么是否有可能通过存储值来丢失数据？我想我是在问是否会出现通常的比赛状况问题。
如果处理器属于不同的子拓扑，会有什么不同吗？
此外，当附加到具有不同数目分区的源的处理器共享相同的状态存储时，会发生什么情况？这将如何影响状态存储分块？
浏览 10提问于2020-06-10得票数 1
回答已采纳
2回答
我们在kotlin中有一个org.apache.kafka:kafka-streams:3.2.3应用程序，并从org.apache.kafka:kafka-streams:3.3.1切换到KStreams，现在在运行时收到以下警告：
WARN org.apache.kafka.streams.internals.metrics.ClientMetrics - Error while loading kafka-streams-version.properties
java.lang.NullPointerException: inStream parameter is null
a
浏览 21提问于2022-10-14得票数 5
我有一个应用程序需要侦听多个不同的主题；每个主题都有处理消息的不同逻辑。我曾想过对每个KafkaStreams实例使用相同的kafka属性，但我得到了一个错误，如下所示。
误差
java.lang.IllegalArgumentException: Assigned partition my-topic-1 for non-subscribed topic regex pattern; subscription pattern is my-other-topic
码 (kotlin)
class KafkaSetup() {
companion object {
pri
浏览 6提问于2017-12-27得票数 15
回答已采纳
当试图从KafkaStreams获取本地状态存储时，如果本地KafkaStreams实例尚未准备好，或者状态存储刚刚迁移到另一个实例(重新平衡)，则可以获得InvalidStateStoreException异常。
假设我们有DSL拓扑，其中包括由addStateStore添加的本地状态存储和一些进程或转换(KStream:process或KStream::transform)。
以下是我的问题：
当从InvalidStateStoreException或Transformer::init内部的ProcessorContext上下文获取本地状态存储时，是否会引发Processor::ini
浏览 1提问于2019-01-18得票数 1
回答已采纳
1回答
目前使用Kafka Streams在客户端系统中聚合事件。当使用假事件运行我们的原型时，一切都运行得很完美。然而，在使用实际数据时，我们注意到在聚合过程中，Streams会自动创建内部主题。虽然在理论上这是很好的，我们的客户有必要的，超级严密的安全性，不愿意授予我的开发团队主题创建特权。这意味着我们不能按原样运行Streams程序。
不过，我们可以为自己创建主题，并使用这些主题来代替创建自己的Kafka主题的Streams。有没有可能/如何开始重定向Streams内部主题创建以利用现有主题？
注意:我们可以随心所欲地命名内部主题。它只需要由拥有这些特权的团队创建。
浏览 21提问于2021-06-08得票数 0
是否可以使用Java CompletableFutures在Kafka stream应用程序中执行并行工作？ 我想从1Kafka主题中读取，创建两个窗口计数，1分钟，另一个小时，但它们是并行的。 我写了一些示例代码。我能够让它工作，但是看一下Kafka流文档，因为KafkaStreams为每个分区分配一个任务，并且它不能超过一个线程，我不确定这个代码是否会有预期的效果。 CompletableFuture completableFutureOfMinute = CompletableFuture.runAsync(() -> {
inputStream.groupByK
浏览 38提问于2020-07-25得票数 0
回答已采纳
我有一个混合匹配的DSL-PAPI拓扑。DSL部件将页面视图(“页面视图”主题)与这些页面视图的用户(“用户”主题)连接起来。我希望两者都加入，因此，如果用户是新的，然后创建一个新的“用户”从pvs信息到“用户”主题，并没有做任何其他事情。
所以我试图在页面视图和用户之间做一个左连接，如果用户为空，这意味着还没有用这个键创建用户，所以我创建了一个。
在代码中，我将页面视图作为流，用户作为表，当用户在联接中出现空值时，将它们连接起来，然后过滤并发送给“用户”这些新用户。
val builder = new StreamsBuilder()
val pageviewsTopic:
浏览 1提问于2018-06-13得票数 0
回答已采纳
我正在学习卡夫卡流，我得到了一个错误，我尝试了一些东西，但是注意到了作品
输入: value_1，value_2，value_3 .
public static void main(String[] args) throws InterruptedException {
String host = "127.0.0.1:9092";
String consumer_group = "firstGroup1";
String topic = "test1";
浏览 16提问于2022-01-08得票数 0
回答已采纳
下面是我的代码片段。我想将kafka流处理并行化。但是我不想放到Runnable中，我也不想多次启动这个应用程序。
有没有像streams.parallel()这样的方法？
final Serde<String> stringSerde = Serdes.String();
Consumed<String, String> types = Consumed.with(stringSerde, stringSerde);
//create StreamFactory
StreamsBuilder bu
浏览 0提问于2018-08-09得票数 1
我一直在探索卡夫卡流的来源，很难理解运营商会为一个或多个子拓扑(然后是)做些什么。有人能解释一下使用什么吗，这样就可以显示子拓扑了吗？
什么是次级拓扑？
浏览 3提问于2021-10-24得票数 1
回答已采纳
我需要在一个春启动应用程序中编写2个处理器。两个流处理器都需要使用来自单主题的消息，并生成输出到另一个单主题。第一个处理器不处理输入主题的所有消息，它处理一些选定的消息(按消息键分组)，并使用windowedBy选项。
然而，第二个处理器需要处理不属于第一个处理器groupBy逻辑的所有消息。
我从第一个处理器开始，它工作得很好。当我尝试介绍第二个处理器时，问题就来了。
KafkaProcessor.java:
@Bean
public Function<KStream<String, Input>, KStream<String, Output>
浏览 9提问于2022-06-08得票数 0
以前我一直在使用0.8API。当您将主题列表传递给它时，它会返回一个流映射(每个主题一个条目)。这允许我生成一个单独的线程，并将每个主题的流分配给它。在每个主题中有太多的数据，产生一个单独的线程有助于多任务处理。
//0.8 code sample
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =
consumer.createMessageStreams(topicCountMap);
我想升级到0.10。我检查了KafkaStreams和KafkaConsumer类。KafkaC
浏览 4提问于2016-09-30得票数 2
回答已采纳
我正在尝试使用__cpuid()收集有关我的CPU的信息。虽然它在我的电脑上运行正常，但当我在同事的电脑上运行我的程序时，它检测到是超线程的，尽管根据英特尔自己网站的规范，它不是超线程的。
__cpuid()还检测到错误数量的“逻辑核心”，如下所示：。它声称有16个逻辑核心和8个物理核心。
我试着在我自己的计算机上运行在那个线程中找到的代码，一个给了我与至强相同的数字。
那么__cpuid()到底有多可靠呢？从我自己的经验来看，它似乎并不那么可靠。我是不是搞错了什么基本的东西？
浏览 0提问于2012-02-14得票数 6
回答已采纳
1回答
关于这个话题有相当多的答案，但都不起作用。
我正在尝试执行以下streams处理器。
object simplestream extends App {
val builder: KStreamBuilder = new KStreamBuilder
val streamingConfig = { //ToDo - Move these to config
val settings = new Properties
settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "example11"
浏览 4提问于2017-08-18得票数 3
回答已采纳
我有一个卡夫卡流应用程序运行(0.10.2.1)。当我关闭Kafka集群时，streams应用程序继续等待下一条消息，当集群被恢复时，它将恢复使用消息。在集群关闭期间，应用程序看起来运行良好。我已经测试了45分钟了。
我希望卡夫卡会抛出一个例外或者停止。我已经将StateListener配置为在KafkaStreams关闭时进行日志记录，但是它从未被调用过。
kafkaStreams.setStateListener((newState, _) => {
if (newState == KafkaStreams.State.NOT_RUNNING) {
Log.error(&
浏览 2提问于2017-06-29得票数 6
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
