清除单元测试的kafka主题

文章/答案/技术大牛

发布

2回答

java、scala、unit-testing、apache-kafka

我需要在避免第三方库的kafka应用程序上执行单元测试。我现在的问题是，我想在测试之间清除所有的主题，但我不知道怎么做。这是我的临时解决方案:提交每次测试后产生的每条消息，并将所有测试消费者放在同一个消费者组中。例如，当我在一个测试中创建一个新的使用者时，messages包含在前一个测试中生成的消息。

浏览 16提问于2016-08-09得票数 9

2回答

使用C#编程清除卡夫卡主题

c#、apache-kafka、confluent-platform、kafka-topic

我需要清除或删除卡夫卡主题编程使用C#语言。目前，我已经使用Confluent.Kafka库发布和消费卡夫卡主题。我可以使用这样的命令行删除Kafka主题是否有使用C#语言以编程方式清除卡夫卡主题的库或方法？

浏览 0提问于2018-12-27得票数 0

回答已采纳

2回答

Kafka broker与“设备上没有剩余空间”

apache-kafka

我有一个6节点的Kafka集群，由于不可预见的情况，其中一个代理上的kafka分区完全填满了。 Kafka understandable不会启动。我们设法处理了来自其他代理的主题的数据。我们的复制因子是4，所以一切都很好。我是否可以手动删除主题中的索引文件，以便kafka可以启动并清除数据本身，或者如果我这样做，是否有损坏的

浏览 68提问于2019-12-20得票数 2

1回答

InvalidStateStoreException:州立商店在卡夫卡流中不开放

apache-kafka、apache-kafka-streams

at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1200) ~[kafka-clients-2.2.1kafka-streams-2.3.0.jar:?][kafka-streams-2.3.0.jar:?]:60)从上面的核心，我在等待商店是可查询的，但我仍然得到商店不开，商

浏览 0提问于2019-07-25得票数 2

回答已采纳

2回答

在将保留时间设置为1秒后，如何验证Kafka主题确实已被清除

apache-kafka

我想清除卡夫卡的一个话题。因此，我将保留时间设置为1秒 /opt/kafka/bin/kafka-configs.sh --zookeeper zk.com --entity-type topics --entity-name/opt/kafka/bin/kafka-configs.sh --zookeeper zk.com --entity-type topics --entity-name my_topi

浏览 0提问于2018-04-10得票数 1

回答已采纳

1回答

如何从字符串创建Kafka Consumer记录以构建Junit测试用例

java、junit、apache-kafka、kafka-consumer-api

我需要一些帮助来为我的Java kafka用户构建一个Junit测试用例。我的原始源代码具有如下方法，并且需要为其创建一个Unit测试用例。我的kafka使用者正在从Kafka主题拉取消息，我需要能够提供ConsumerRecords格式的输入消息，但作为单元测试的一部分，我不会从kafka中轮询消息，而是模拟原始kafka主题中的消息，并将静

浏览 0提问于2018-05-02得票数 2

1回答

墓碑和钥匙店的清理

apache-kafka、kafka-consumer-api、apache-kafka-streams

我有一些用Java实现的Kafka用户，我正在实现一个独立的应用程序来检查记录和墓碑。希望卡夫卡将删除州商店，因为它压缩主题。卡夫卡删除相应主题中的旧记录时，是否删除了？当您在相应的主题中删除墓碑记录时，删除了吗？，我们是否坚持了？changelogKSTREAM-KEY-SELECT重新分区对于使用聚合函数

浏览 4提问于2020-02-26得票数 2

1回答

kafka连接器更新rest api不更新接收器数据

mysql、apache-kafka、apache-kafka-connect、debezium

我正在使用错误的转换运行接收器，这会在接收器mysql表中产生错误的数据，但当我使用PUT rest API更正接收器配置并运行它时，它运行成功但没有更新接收器表中的现有数据？

浏览 0提问于2019-08-23得票数 2

1回答

如何在清除kafka主题后返回原始日志保留

linux、kafka

我们有3台卡夫卡机器版本- 0.10.0.2.6kafka-topics.sh --zookeeper localhost:2181 --alter --topic Topic1 --config retention.ms=1000 kafka-topics.sh --zookeeper localhost:2181 --alter --topic To

浏览 0提问于2018-05-16得票数 0

回答已采纳

1回答

卡夫卡溪流-洪水卡夫卡原木

apache-kafka、apache-kafka-streams

我正在使用Kafka流，我注意到它使我的kafka日志记录了很多日志消息，例如： [2019-04-17 09:06:16,541] INFO [Log partition=my-application-KSTREAM-AGGREGATE-STATE-STORE-0000000076-repartition-0, dir=/opt/kafka/data/logs] Incrementing log start offset to 316423 (kafka.log.

浏览 1提问于2019-04-17得票数 3

1回答

卡夫卡+如何计算log.retention.byte值

apache-kafka

log.retention.byte参数的主要作用之一是避免卡夫卡磁盘的完整大小，或者换句话说，清除数据日志以避免卡夫卡磁盘满。为了更好地理解它，让我们举个小例子(实际操作总是要好得多)/var/kafka/ka

浏览 1提问于2018-10-29得票数 12

回答已采纳

2回答

当Spark消费者读取时，Kafka主题不会被刷新

apache-spark、apache-kafka、spark-structured-streaming

我在Spark中使用以下消费者代码来读取Kafka主题： .read .option("kafka.bootstrap.servers.load() .as[(String, String)] 代码按预期从主题中读取，但主题</

浏览 1提问于2019-01-12得票数 0

1回答

如何清除信息中心的主题？

apache-kafka、ibm-cloud、message-hub

在使用MessageHub进行开发时，我经常发现我想从某个主题中清除我的开发数据。如何清除MessageHub主题？这个问题类似于，但是不同，因为这个问题是针对apache的，我不确定Message是否支持kafka命令行工具。

浏览 3提问于2016-12-05得票数 2

回答已采纳

1回答

Kafka Topic retention.ms在Spark Streaming上下文中使用时不工作

apache-kafka、spark-streaming

我正在运行Spark Streaming作业(意味着数据不断被推送到kafka主题，并被Spark消费者持续读取)。我的Kafka输入数据主题的保留时间设置为60000 (1分钟)。但是，Input Topic在1分钟后不会清除消息。如果没有新数据添加到主题中，则大约需要1分26分钟的时间来清除。如果我连续添加数据两分钟，由于retention.ms设置为1分钟，我预计将清除一半的

浏览 0提问于2015-05-01得票数 0

1回答

卡夫卡消费者/生产者性能测试

performance、stress-testing

我开发了一个消费者卡夫卡，它做了以下工作：-得到一个关于卡夫卡主题的事件。-执行一些操作-发布一个或多个关于卡夫卡主题的事件我正在寻找一种解决方案，使用Jmeter或Gateling。

浏览 0提问于2019-12-30得票数 1

1回答

KStreams正在尝试删除一个重分区主题。

apache-kafka、apache-kafka-streams

我们已经构建了一个从主题读取并在不同字段上执行groupBy的管道。然而，KStream不断地向Kafka发送Delete请求。(cached) (io.aiven.kafka.auth.AivenAclAuthorizer)更新就我在KStreams源代码上的跟踪而言，它在中调用了KafkaAdminClien

浏览 5提问于2019-12-11得票数 2

回答已采纳

1回答

卡夫卡流TopologyTestDriver输入输出主题

apache-kafka-streams

我有卡夫卡流单元测试，基于一个非常伟大、可靠和方便的TopologyTestDriver：org.apache.kafka.streams.errors.TopologyException

浏览 3提问于2020-05-03得票数 0

1回答

清除检查点位置后Spark结构化流的旧Kafka偏移消费

java、apache-spark、apache-kafka、spark-structured-streaming

场景：在清除检查点位置之后，我期待的只是流上的新消息。星火版本: 2.4.0，Kafka -客户端版本: 2.0.0，Kafka版本: 2.0.0

浏览 0提问于2019-01-17得票数 1

回答已采纳

2回答

应用程序重新启动后，GlobalKTable是否会保留数据？

apache-kafka、apache-kafka-streams、spring-kafka、spring-cloud-stream

我正在使用Spring Cloud Streams，我有一个接收KStream和GlobalKTable的BiFunction。我不想在应用程序重新启动后丢失GlobalKTable数据，但这正是正在发生的事情。MyClass2>, KStream<String, MyClass3>> process() { }spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.mat

浏览 3提问于2021-05-14得票数 0

1回答

kafka在哪里保存本地状态存储？

apache-kafka、apache-kafka-streams

我创建了一个Kafka主题，并向它发送了一些消息。我看到新目录已在我配置的state文件夹中创建。我可以从当地的州立商店里读到这笔钱。然后，我关闭了正在运行流的应用程序。我重启Kafka集群。我重新启动了具有流拓扑的应用程序。在我的理解中，状态已经消失了。K

浏览 0提问于2021-04-12得票数 0

点击加载更多