Kafka Producer无法推送后续事务批次消息 报-172状态非法错误

你遇到的报错和后续批次发送失败问题来自三个典型的Kafka事务生产者使用错误：

• 重复调用init_transactions()：该API是生产者生命周期内仅允许执行1次的初始化方法，作用是向Broker注册事务生产者身份、完成事务上下文初始化，执行完成后生产者会从初始状态进入Ready状态。你把该方法放在每次批量发送的send_in_transaction函数中循环调用，首次调用成功后，第二次触发时生产者已经处于Ready状态，就会直接抛出KafkaError{code=_STATE,val=-172,str="Operation not valid in state Ready"}错误，这是报错的直接原因。
• transactional.id配置不符合规范：事务ID是Broker侧识别同一逻辑事务生产者的唯一标识，同一生产者实例生命周期内需要使用固定值，不能通过随机uuid动态生成。动态生成事务ID会导致Broker无法关联历史事务状态，触发事务僵尸拦截、消息乱序、事务失效等问题。
• 缺失客户端事件轮询逻辑：confluent-kafka客户端采用异步设计，produce()方法仅将消息暂存在本地内存队列，不会直接触发网络发送。你没有在生产消息后调用poll()方法处理后台IO事件、发送结果回调，本地队列会逐渐被占满，就算关闭事务机制，后续批次的消息也无法写入队列完成投递。

按照以下步骤调整代码即可，不需要为每个批次新建生产者实例：

1. 调整生产者初始化逻辑：将init_transactions()的调用移到生产者实例创建完成后，全局仅执行1次，不要放在单次发送逻辑中。
2. 修正transactional.id配置：替换动态生成的uuid为固定的业务标识（例如biz_name_batch_event_producer），多实例部署时可追加实例唯一后缀，保证每个运行中的生产者实例事务ID全局唯一即可。
3. 补充轮询逻辑：在消息生产、事务提交/中止后调用非阻塞的poll(0)方法，让客户端及时处理后台事件、清理已完成的消息队列。

修正后的代码示例如下：

from confluent_kafka.avro import AvroProducer
from confluent_kafka import KafkaException
from confluent_kafka.avro.serializer import AvroTypeException
import logging
logger = logging.getLogger(__name__)

BOOTSTRAP_SERVERS = "你的Kafka集群地址"
SCHEMA_REGISTRY_URL = "你的Schema Registry地址"
value_schema = "业务对应的Avro Schema定义"

DEFAULT_PRODUCER_CONFIG = {
    "bootstrap.servers": BOOTSTRAP_SERVERS,
    "schema.registry.url": SCHEMA_REGISTRY_URL,
    "enable.idempotence": True,
    "retries": 3,
    "acks": "all",
    "security.protocol": "SSL",
    # 替换为固定的业务标识，不要动态生成随机值
    "transactional.id": "user_behavior_event_producer_v1"
}

# 全局复用生产者实例，不要每次发消息新建
producer = AvroProducer(
    DEFAULT_PRODUCER_CONFIG, default_value_schema=avro.loads(value_schema)
)
# 事务初始化仅执行1次
producer.init_transactions()

修正后的批量事务发送方法：

def send_in_transaction(producer, topic, events):
    producer.begin_transaction()
    try:
        for event in events:
            producer.produce(topic=topic, key=None, value=event)
            # 非阻塞轮询，处理已就绪的后台IO事件
            producer.poll(0)
        # 等待事务提交完成，超时时间60秒
        producer.commit_transaction(60)
        producer.poll(0)
    except AvroTypeException as ex:
        producer.abort_transaction()
        producer.poll(0)
        logger.error(f"Avro serialization error: {str(ex)}")
        raise KafkaException("Error sending events to kafka")
    except Exception as ex:
        producer.abort_transaction()
        producer.poll(0)
        logger.error(f"Error sending events to kafka: {str(ex)}")
        raise KafkaException("Error sending events to kafka")

• 你之前为每笔事务新建生产者实例的方式能跑通，本质是因为每个新实例只调用了一次init_transactions()，但生产者是重量级线程安全对象，频繁创建销毁会产生大量冗余TCP连接、给Broker增加不必要的事务元数据存储压力，生产环境不推荐使用。
• 服务优雅退出前，调用一次producer.flush()等待本地队列所有消息完成投递即可，避免消息丢失。

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