apache-kafka - 卡夫卡流 : PolicyViolationException: Topic replication factor must be 3-6ren

apache-kafka - 卡夫卡流 : PolicyViolationException: Topic replication factor must be 3

转载作者：行者123 更新时间：2023-12-04 15:34:25

我目前正在构建一个应用程序，它写入一个 kafka 主题并监听同一个主题以从中生成一个 ktable 并将其具体化到一个商店中。我正在运行的代码基于以下 sample .我几乎复制了其中的大部分内容(除了 PageViewEventSource 之外的所有内容)并将名称重构到我的用例中。我还使用示例中使用的 key 更新了我的 application.properties。

运行应用程序时出现以下错误:

2020-02-12 17:54:31.982 ERROR 69005 --- [-StreamThread-1] o.a.k.s.p.i.InternalTopicManager         : stream-thread [restartedMain] Unexpected error during topic creation for pairing-events-pcmv-changelog.
Error message was: org.apache.kafka.common.errors.PolicyViolationException: Topic replication factor must be 3
2020-02-12 17:54:31.986 ERROR 69005 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1] Encountered the following unexpected Kafka exception during processing, this usually indicate Streams internal errors:

org.apache.kafka.streams.errors.StreamsException: Could not create topic pairing-events-pcmv-changelog.
    at org.apache.kafka.streams.processor.internals.InternalTopicManager.makeReady(InternalTopicManager.java:148) ~[kafka-streams-2.3.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.prepareTopic(StreamsPartitionAssignor.java:1049) ~[kafka-streams-2.3.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.assign(StreamsPartitionAssignor.java:635) ~[kafka-streams-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.performAssignment(ConsumerCoordinator.java:424) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.onJoinLeader(AbstractCoordinator.java:622) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.access$1100(AbstractCoordinator.java:107) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:544) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:527) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:978) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:958) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:578) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:388) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:294) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:415) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:358) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:353) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1251) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1201) ~[kafka-clients-2.3.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:963) ~[kafka-streams-2.3.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:859) ~[kafka-streams-2.3.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:819) ~[kafka-streams-2.3.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:788) ~[kafka-streams-2.3.1.jar:na]
Caused by: org.apache.kafka.common.errors.PolicyViolationException: Topic replication factor must be 3

2020-02-12 17:54:31.986  INFO 69005 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1] State transition from PARTITIONS_REVOKED to PENDING_SHUTDOWN
2020-02-12 17:54:31.986  INFO 69005 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1] Shutting down
2020-02-12 17:54:31.986  INFO 69005 --- [-StreamThread-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions
2020-02-12 17:54:31.986  INFO 69005 --- [-StreamThread-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1-producer] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
2020-02-12 17:54:31.991  INFO 69005 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD
2020-02-12 17:54:31.991  INFO 69005 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c] State transition from REBALANCING to ERROR
2020-02-12 17:54:31.991 ERROR 69005 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c] All stream threads have died. The instance will be in error state and should be closed.
2020-02-12 17:54:31.991  INFO 69005 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1] Shutdown complete
Exception in thread "pairing-events-2d949c85-df82-4961-a849-6c6e079edf9c-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Could not create topic pairing-events-pcmv-changelog.
    at org.apache.kafka.streams.processor.internals.InternalTopicManager.makeReady(InternalTopicManager.java:148)
    at org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.prepareTopic(StreamsPartitionAssignor.java:1049)
    at org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.assign(StreamsPartitionAssignor.java:635)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.performAssignment(ConsumerCoordinator.java:424)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.onJoinLeader(AbstractCoordinator.java:622)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.access$1100(AbstractCoordinator.java:107)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:544)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:527)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:978)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:958)
    at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204)
    at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167)
    at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:578)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:388)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:294)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:415)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:358)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:353)
    at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1251)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1201)
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:963)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:859)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:819)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:788)
Caused by: org.apache.kafka.common.errors.PolicyViolationException: Topic replication factor must be 3

任何修复方法的线索？

最佳答案

您的代理设置要求最小复制因子为 3。

您可以为绑定(bind)设置 ... topic.replication-factor 属性。

参见 Consumer Properties在 Binder 文档中。

关于apache-kafka - 卡夫卡流 : PolicyViolationException: Topic replication factor must be 3，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/60194656/

文章推荐： ruby-on-rails - 如何悲观地锁定Rails中的多行？

文章推荐： slurm - Slurm 中的默认作业时间限制

文章推荐： css - 背景图片 CSS3 无限循环动画

factor-lang - 良好的学习资源 Factor
就目前而言，这个问题不适合我们的问答形式。我们希望答案得到事实、引用或专业知识的支持，但这个问题可能会引起辩论、争论、投票或扩展讨论。如果您觉得这个问题可以改进并可能重新打开，visit the he
factor-lang - 您打算使用 Factor 编程语言吗？
就目前而言，这个问题不适合我们的问答形式。我们希望答案得到事实、引用或专业知识的支持，但这个问题可能会引起辩论、争论、投票或扩展讨论。如果您觉得这个问题可以改进并可能重新打开，visit the he
factor-lang - 将引用保留为 Factor 中的元组成员
我想将引用保留为 Factor 中元组的成员。但是，当我尝试对其执行“调用”时，出现错误“无法将调用应用于运行时计算值”。请注意，将函数标记为“内联”不会执行任何操作。示例代码: USING: ac
r - 为什么使用 as.factor() 而不仅仅是 factor()
我最近看到 Matt Dowle 用 as.factor() 写了一些代码，具体来说 for (col in names_factors) set(dt, j=col, value=as.facto
r - dplyr 和 group_by : factor vs no factor
(所描述的行为是一个错误!) 我不明白 group_by 对作为因素的列和不是因素的列的影响。下面分析这两种可能性: library(dplyr) df % group_by(height) %>%
factor-lang - 使用 Factor 构建 Web 应用程序
有没有人用 Factor 构建了一个 Web 应用程序？ ?您在此过程中遇到了哪些绊脚石或问题？最佳答案试试 this . 关于factor-lang - 使用 Factor 构建 Web 应用程
akka - 配置 core-pool-size-factor 和 max-pool-size-factor 设置的一般好做法是什么？
例如，如果 CPU 有四个内核和八个内核线程。我应该设置核心池大小因子一直设置到8？的一般尺寸是多少？最大池大小因子关于核心池大小因子 ? 我提到的其他设置是否与 Akka 配置相关？最
r - 尝试使用 "bnlearn"实现一个简单的朴素贝叶斯分类器。不断出现错误 "variables must be either numeric, factors or ordered factors"
我正在尝试通过重新创建给我的数据结果来在 R 中实现 NB 分类器。现在我只是对训练数据本身进行测试，看看准确性如何。数据集中有 29 个变量，其中一个称为“状态”。它有两个值:Win 和 Lose
factorization - 关于两个数之间关系的问题
当一个可被另一个整除时，数字的位之间是否存在任何关系？ 36位与9位或4位或12位、10位(1010)与5位(101)、21位(10101)与7位(00111)的位序列有什么关系？谢谢。如果有些句子
prime-factoring - 如何在没有除法的线性筛算法中寻找一个整数的因式分解？
我学会了一种叫做“线性筛”的算法https://cp-algorithms.com/algebra/prime-sieve-linear.html能够在线性时间内得到所有小于 N 的素数。这个算法有
r - factor() 中的未指定水平
我正在使用 R 中的一个数据集，它带有一个代码本，它基本上告诉我因子变量的不同级别的标签应该是什么。例如，使用密码本，我可以看到在我的“性别”变量中，0 是“女性”，1 是“男性”。我正在使用此信息相
R sapply is.factor
我试图将数据集分成具有因子变量和非因子变量的部分。我正在做类似的事情: 这部分工作: factorCols <- sapply(df1, is.factor) factorDf <- df1[,fa
R:为什么我在将列转换为因子后没有得到类型或类 "factor"？
我有以下设置。 df <- data.frame(aa = rnorm(1000), bb = rnorm(1000)) apply(df, 2, typeof) # aa bb
r - as.factor 不将整数转换为因子
我现在正在自学 R。我正在尝试使用以下内容将整数变量转换为分类变量。 train[, c("Store", "DayOfWeek")] str(mtcars) 'data.frame': 32
factor-lang - 如何对一个数进行因数立方？
我正在玩 Factor，试图对串联编程有一点了解。写一个词来平方一个数字是微不足道的: : square ( n -- n ) dup * ; 但对于我来说，我似乎无法弄清楚如何计算一个数字的立方:
factors - 找出一个数的两个因数，它们彼此相似或接近
给定一个数 x，我如何找到两个数 y 和 z，这样 x = y * z 和 y==Z 或者 y 和 z 彼此接近？此外，x、y、z 都是整数。示例: x = 16484, y=z=128; x=
R:为什么我在将列转换为因子后没有得到类型或类 "factor"？
我有以下设置。 df <- data.frame(aa = rnorm(1000), bb = rnorm(1000)) apply(df, 2, typeof) # aa bb
javascript - Factors 函数获取最小公倍数或素数
关闭。这个问题需要多问focused 。目前不接受答案。想要改进此问题吗？更新问题，使其仅关注一个问题 editing this post . 已关闭 3 年前。 Improve this ques
斯普伦克 : How to figure out replication Factor
如果你觉得这听起来很傻，我提前道歉，我是 splunk 的新手，并且学过 udemy 类(class)，但无法弄清楚这一点。 If I check my indexes.conf file in cl
r - 为什么在unicode字符串上的as.factor()对于每个操作系统都返回不同的结果？
为什么此代码:as.factor(c("\U201C", '"3', "1", "2", "\U00B5"))，在每个操作系统上返回因子级别的不同顺序？在Linux上: > as.factor(c(

行者123

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

apache-kafka - 卡夫卡流 : PolicyViolationException: Topic replication factor must be 3