在执行kstreams以获取聚合计数时,我将得到此错误。
Exception in thread "KStreamThread-1" org.apache.kafka.streams.errors.ProcessorStateException: task [1_2] Failed to flush state store KSTREAM-AGGREGATE-STATE-STORE-0000000002
at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorSta
当使用处理器API和KTable函数创建全局addGlobalStore时,结果存储将填充OK。但是,随后对存储内容进行迭代的尝试会导致以下异常:
Exception in thread "main" java.lang.NullPointerException at org.apache.kafka.streams.state.internals.Serializ
我有一个处理器,它从主题中提取json字符串,并输入GenericRecord。现在我把小溪分成两条树枝。我接受第一个分支,并将( key,value)映射为包含json特定字段和该字段的值的2个字符串中的(key,value ),并按键分组。到目前为止还不错。现在,我必须使用新定义的Type用户聚合流,并收到一个异常。
在这里,代码:
新型:
private class Tuple {
public int occ;
public int sum;
public Tuple (int occ, int sum) {
this.occ = occ;
KafkaStream在很长一段时间后无法生成数据。(超过设定的过期时间)
在记录错误消息后,甚至KafkaStream也死了。
例外情况如下:
org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:784)
at o
我正在寻找一些Kafka流的汇合示例,配置值'StreamsConfig.COMMIT_INTERVAL_MS_CONFIG‘的不同值让我有点困惑。 例如,在微服务示例中, config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1); //commit as fast as possible https://github.com/confluentinc/kafka-streams-examples/blob/5.1.0-post/src/main/java/io/confluent/examples/streams/microser
启动流应用程序(使用卡夫卡流)失败时使用"java.lang.IllegalStateException:这不应该发生,因为headers()应该只在处理记录时调用“
这似乎只有当我启动应用程序时,在主题中已经有数据时才会发生。如果主题是空的,并且我开始将数据推送给它,那么一切都很好。
有人知道为什么会发生这种事吗?
谢谢
This should not happen as headers() should only be called while a record is processed
java.lang.IllegalStateException: This should no
基本上我有记账课。我有数据。我想将这些对象发送到我与生产者的主题中。现在还可以。稍后,我想使用Kafka流进行聚合,但我不能这样做,因为有些Serde属性在我的配置中是错误的,我认为:/。我不知道错误在哪里。我的制片人工作很好,但我不能汇总。有人帮我看看我的卡夫卡流代码吗?我的帐户类:
public class Account {
private long fromId;
private long amount;
private long toId;
private ZonedDateTime time;
}
我的帐户类有两个类序列化器和反序列化器。序列化器:
当源主题分区计数= 1时工作正常。如果我将分区增加到任何大于1的值,我会看到下面的错误。既适用于低级,也适用于DSL API。有什么建议吗?可能会遗漏什么?
org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:410)
at org.apach
当我使用聚合函数模拟单词计数时,我会遇到Serde铸造问题。
Exception in thread "aggregation-transformation-application-43485635-2d3c-4edc-b13c-c6505a793d18-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather