如何解决Debezium-MySQL连接器的Kafka NOT_ENOUGH_REPLICAS警告？

我们的Debezium-MySQL连接器日志中每秒出现10次以下警告：

[kafka-producer-network-thread | connect-distributed-offsets] 
WARN  org.apache.kafka.clients.producer.internals.Sender - 
[Producer clientId=connect-distributed-offsets] Got error produce response with 
correlation id 34626 on topic-partition 
debezium-events-offset-dev-topic-events2-0, retrying (2147449048 attempts left). 
Error: NOT_ENOUGH_REPLICAS

生产者端配置： #

org.apache.kafka.clients.producer.ProducerConfig - Idempotence will be disabled because acks is set to 1, not set to 'all'." 
org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values:" 
"acks = 1" 
"auto.include.jmx.reporter = true" 
"batch.size = 32768" 
"bootstrap.servers = [b-1.eventsmskafka.qgfgzh.c14.kafka.us-east-1.amazonaws.com:9092, b-2.eventsmskafka.qgfgzh.c14.kafka.us-east-1.amazonaws.com:9092, b-3.eventsmskafka.qgfgzh.c14.kafka.us-east-1.amazonaws.com:9092]" 
"buffer.memory = 1048576" 
"client.dns.lookup = use_all_dns_ips" 
"client.id = debezium-cdc-events2-schemahistory" 
"compression.type = none" 
"connections.max.idle.ms = 540000" 
"delivery.timeout.ms = 120000" 
"enable.idempotence = false" 
"interceptor.classes = []" 
"key.serializer = class org.apache.kafka.common.serialization.StringSerializer" 
"linger.ms = 0" 
"max.block.ms = 10000" 
"max.in.flight.requests.per.connection = 5" 
"max.request.size = 1048576" 
"metadata.max.age.ms = 300000" 
"metadata.max.idle.ms = 300000" 
"metric.reporters = []" 
"metrics.num.samples = 2" 
"metrics.recording.level = INFO" 
"metrics.sample.window.ms = 30000" 
"partitioner.adaptive.partitioning.enable = true" 
"partitioner.availability.timeout.ms = 0" 
"partitioner.class = null" 
"partitioner.ignore.keys = false" 
"receive.buffer.bytes = 32768" 
"reconnect.backoff.max.ms = 1000" 
"reconnect.backoff.ms = 50" 
"request.timeout.ms = 30000" 
"retries = 1" 
"retry.backoff.ms = 100" 
"sasl.client.callback.handler.class = null" 
"sasl.jaas.config = null" 
"sasl.kerberos.kinit.cmd = /usr/bin/kinit" 
"sasl.kerberos.min.time.before.relogin = 60000" 
"sasl.kerberos.service.name = null" 
"sasl.kerberos.ticket.renew.jitter = 0.05" 
"sasl.kerberos.ticket.renew.window.factor = 0.8" 
"sasl.login.callback.handler.class = null" 
"sasl.login.class = null" 
"sasl.login.connect.timeout.ms = null" 
"sasl.login.read.timeout.ms = null" 
"sasl.login.refresh.buffer.seconds = 300" 
"sasl.login.refresh.min.period.seconds = 60" 
"sasl.login.refresh.window.factor = 0.8" 
"sasl.login.refresh.window.jitter = 0.05" 
"sasl.login.retry.backoff.max.ms = 10000" 
"sasl.login.retry.backoff.ms = 100" 
"sasl.mechanism = GSSAPI" 
"sasl.oauthbearer.clock.skew.seconds = 30" 
"sasl.oauthbearer.expected.audience = null" 
"sasl.oauthbearer.expected.issuer = null" 
"sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000" 
"sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000" 
"sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100" 
"sasl.oauthbearer.jwks.endpoint.url = null" 
"sasl.oauthbearer.scope.claim.name = scope" 
"sasl.oauthbearer.sub.claim.name = sub" 
"sasl.oauthbearer.token.endpoint.url = null" 
"security.protocol = PLAINTEXT" 
"security.providers = null" 
"send.buffer.bytes = 131072" 
"socket.connection.setup.timeout.max.ms = 30000" 
"socket.connection.setup.timeout.ms = 10000" 
"ssl.cipher.suites = null" 
"ssl.enabled.protocols = [TLSv1.2, TLSv1.3]" 
"ssl.endpoint.identification.algorithm = https" 
"ssl.engine.factory.class = null" 
"ssl.key.password = null" 
"ssl.keymanager.algorithm = SunX509" 
"ssl.keystore.certificate.chain = null" 
"ssl.keystore.key = null" 
"ssl.keystore.location = null" 
"ssl.keystore.password = null" 
"ssl.keystore.type = JKS" 
"ssl.protocol = TLSv1.3" 
"ssl.provider = null" 
"ssl.secure.random.implementation = null" 
"ssl.trustmanager.algorithm = PKIX" 
"ssl.truststore.certificates = null" 
"ssl.truststore.location = null" 
"ssl.truststore.password = null" 
"ssl.truststore.type = JKS" 
"transaction.timeout.ms = 60000" 
"transactional.id = null" 
"value.serializer = class org.apache.kafka.common.serialization.StringSerializer" 

解决方案： #

1. 检查Kafka主题的副本与ISR状态

• 执行以下命令查看目标主题的详细状态：

  kafka-topics.sh --describe --topic debezium-events-offset-dev-topic-events2 --bootstrap-server b-1.eventsmskafka.qgfgzh.c14.kafka.us-east-1.amazonaws.com:9092
  

• 重点关注ReplicationFactor和ISR列：如果ISR集合中的副本数小于配置的副本数，说明存在未同步的副本。此时需要检查对应Broker的运行状态，修复故障节点后等待副本重新同步。

2. 调整生产者可靠性配置

• 优化重试策略：当前retries=1，可适当提高重试次数（如设置为10），同时调整retry.backoff.ms（如设为200），减少频繁重试的日志刷屏，同时提升写入成功率：

  retries=10
  retry.backoff.ms=200
  

• 调整acks配置：如果业务允许牺牲部分吞吐量换取更高可靠性，可将acks从1改为all，同时确保主题的min.insync.replicas配置与副本数匹配（例如副本数为3时，min.insync.replicas=2），避免因单个Broker故障导致写入失败。

3. 调整主题的副本配置

• 如果目标主题的副本数过低（如仅为1），执行以下命令增加副本数（示例改为3）：

  kafka-topics.sh --alter --topic debezium-events-offset-dev-topic-events2 --replication-factor 3 --bootstrap-server b-1.eventsmskafka.qgfgzh.c14.kafka.us-east-1.amazonaws.com:9092
  

• 同时设置主题级别的min.insync.replicas：

  kafka-configs.sh --alter --topic debezium-events-offset-dev-topic-events2 --add-config min.insync.replicas=2 --bootstrap-server b-1.eventsmskafka.qgfgzh.c14.kafka.us-east-1.amazonaws.com:9092
  

4. 检查Debezium专用主题的配置

Debezium的offset、schema history等主题（如connect-distributed-offsets、debezium-cdc-events2-schemahistory）的可用性直接影响连接器运行，需确保这些主题的副本数和ISR状态正常，若存在同样问题，按上述步骤修复。

内容的提问来源于stack exchange，提问作者user674669