标签: kafka-topic

我已经按照https://kafka.apache.org/24/documentation/streams/developer-guide/testing.html为 kafka 流应用程序编写了一个测试类 ，其代码是

import com.EventSerde;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.Properties;

public class KafkaStreamsConfigTest {

private TopologyTestDriver testDriver;
private TestInputTopic<String, Object> inputTopic;
private TestOutputTopic<String, Object> outputTopic;

private Serde<String> stringSerde = new Serdes.StringSerde();
private EventSerde eventSerde= new EventSerde();

private String key="test";
private Object value = "some value";
private Object expected_value = "real value";

String kafkaEventSourceTopic = "raw_events";
String kafkaEventSinkTopic = "processed_events";
String kafkaCacheSinkTopic = "cache_objects";

String applicationId = "my-app";
String …

在许多文章中，我读到压缩的 Kafka 主题可以用作数据库。但是，在查看 Kafka API 时，我找不到允许我根据键查询主题的值的方法。

那么，压缩后的Kafka主题可以用作（高性能、只读）键值数据库吗？

在我的架构中，我想为组件提供一个紧凑的主题。我想知道该组件是否需要在其本地数据库中拥有该主题的副本，或者是否可以使用该压缩主题作为键值数据库。

有如下文件所述，可以在 Kafka 集群仍在运行时进行 Kafka 主题级别的配置更改。

• https://docs.confluent.io/current/kafka/dynamic-config.html#changed-topic-configurations-dynamically

我的问题是：

• 配置更改何时生效？是否需要重启Kafka集群？ https://community.cloudera.com/t5/Support-Questions/Kafka-alter-topic-requires-broker-restart-to-take-effect/td-p/44396

• 在不同的主题级别配置中，配置更改效果是否存在差异？http://kafka.apache.org/documentation/#topicconfigs

我的 kafka 集群有 3 个代理和几个主题，每个主题有 5 个分区。现在我想为分区设置复制因子。

我可以为 kafka 主题的分区设置的最大复制因子是多少？

I am exploring different PubSub platforms and I was wondering what the limits are in Kafka for listening to multiple topics. Consider for instance this Use Case. We have trains, station entry gates, devices that all publish their telemetry. Currently this is done on a MQ but as data rates increase, smart trains etc. we need to move to a new PubSub/streaming platform and Kafka is on that list of course.

As I see it there are two strategies for …

kafka 领导者是自己分区还是经纪人？我最初的理解是，它们是充当读/写代理的分区，然后将它们的价值交给 ISR。

但是最近我听到他们提到他们好像发生在“经纪人”级别，因此我很困惑。

我知道还有其他帖子旨在回答这个问题，但那里的答案没有帮助。

日志压缩主题不应该针对相同的键保留重复项。但在我们的例子中，当发送具有相同键的新值时，不会删除前一个值。可能是什么问题？

val TestCompactState: KTable[String, TestCompact] = builder.table[String, TestCompact](kafkaStreamConfigs.getString("testcompact-source"),
   (TestCompactmaterialized).withKeySerde(stringSerde).withValueSerde(TestCompactSerde)) 

我得到的 实际结果

Offsets      Keys        Messages
5            {"id":5}   {"id":5,"namee":"omer","__deleted":"false"}
6            {"id":5}   {"id":5,"namee":"d","__deleted":"false"}

我只想要针对相同关键预期结果的最新记录

6            {"id":5}   {"id":5,"namee":"d","__deleted":"false"}