You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by "zhuming (Jira)" <ji...@apache.org> on 2025/10/15 06:46:00 UTC
[jira] [Resolved] (KAFKA-19788) kafka server ClusterAuthorization don't support acks=-1 for kafka-client version>3.1.0
[ https://issues.apache.org/jira/browse/KAFKA-19788?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
zhuming resolved KAFKA-19788.
-----------------------------
Resolution: Fixed
> kafka server ClusterAuthorization don't support acks=-1 for kafka-client version>3.1.0
> --------------------------------------------------------------------------------------
>
> Key: KAFKA-19788
> URL: https://issues.apache.org/jira/browse/KAFKA-19788
> Project: Kafka
> Issue Type: Bug
> Components: producer
> Affects Versions: 3.1.1
> Reporter: zhuming
> Priority: Major
>
> * kafka server version is 2.5.1
> * kafka-client version bigger than 3.1.1
>
> {code:java}
> import org.apache.kafka.clients.producer.KafkaProducer;
> import org.apache.kafka.clients.producer.Producer;
> import org.apache.kafka.clients.producer.ProducerRecord;
> import java.util.Properties;
> public class KafkaSendTest {
> public static void main(String[] args) {
> Properties props = new Properties();
> props.put("bootstrap.servers", "xxx:9092,xxxx:9092");
> props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
> props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
> props.put("transaction.timeout.ms", "300000");
> props.put("acks", "-1"); // If acks=1 or acks=0 it will send successfully
> props.put("compression.type", "lz4");
> props.put("security.protocol", "SASL_PLAINTEXT");
> props.put("sasl.mechanism", "SCRAM-SHA-256");
> props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"xxx\" password=\"xxxx\";");
> Producer<byte[], byte[]> producer = new KafkaProducer<>(props);
> try {
> String topic = "topic1";
> byte[] value = new byte[]{1,2}; // example
> ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(topic, null, value);
> producer.send(record, (metadata, exception) -> {
> if (exception == null) {
> System.out.printf("Sent record(key=%s value=%s) meta(partition=%d, offset=%d)\n",
> record.key(), new String(record.value()), metadata.partition(), metadata.offset());
> } else {
> exception.printStackTrace();
> }
> });
> producer.close();
> } catch (Exception e) {
> e.printStackTrace();
> }
> }
> } {code}
> pom.xml config
> {code:java}
> <dependency>
> <groupId>org.apache.kafka</groupId>
> <artifactId>kafka-clients</artifactId>
> <version>3.4.0</version>
> </dependency> {code}
> When kafka producer acks=-1, It will throw exception.
>
> {code:java}
> org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:1010) at org.apache.kafka.clients.producer.internals.TransactionManager.maybeAddPartition(TransactionManager.java:328) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:1061) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:962) at com.mvad.realtime.show.converter.DataEntryToRTLogConverterTest.main(DataEntryToRTLogConverterTest.java:34)Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed. {code}
> If acks=1 or acks=0 it will send successfully
> {code:java}
> Sent record(key=null ) meta(partition=6, offset=321496) {code}
> acks=-1 is just a param, How it effects ClusterAuthorization of kafka producer.
> Is this a bug or a mechanism in itself?
>
> If change kafka-client verison to 3.1.0. When kafka producer acks=-1, It will send successfully
> {code:java}
> <dependency>
> <groupId>org.apache.kafka</groupId>
> <artifactId>kafka-clients</artifactId>
> <version>3.1.0</version>
> </dependency> {code}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)