You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Abhishek Verma <ab...@gmail.com> on 2017/11/02 09:15:51 UTC
Reg. Kafka transactional producer and consumer
Hi All,
I am trying to make a hello world example for Transactional Producer and trying to consume.
I am doing all this in plain java.
I can produce but consumer is not consuming message.
I searched over other places and I found some people have same problem.
Right now, I am using single broker. I tried same with 3 brokers also and it was not working at that time also.
I don’t know what I am missing and where… :p in Consumer I am missing something or in producer.
I have attached Producer and Consumer codes and console logs with my broker logs
Thanks,
Abhishek
My Broker logs after producing messages
<LOGS broker>
[2017-11-01 18:45:55,000] INFO Updated PartitionLeaderEpoch. New: {epoch:4, offset:3}, Current: {epoch:3, offset0} for Partition: __transaction_state-2. Cache now contains 1 entries. (kafka.server.epoch.LeaderEpochFileCache)
[2017-11-01 18:46:03,482] INFO [Transaction Coordinator 1001]: Initialized transactionalId TXN_ID:0.5031925219291776-156417066 with producerId 4001 and producer epoch 0 on partition __transaction_state-2 (kafka.coordinator.transaction.TransactionCoordinator)
</LOGS broker>
My producer code is
<CODE producer>
import com.example.transaction.producer.utils.DataObject;
import com.example.transaction.producer.utils.serde.JsonSerializer;
import kafka.common.AuthorizationException;
import kafka.common.KafkaException;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.OutOfOrderSequenceException;
import org.apache.kafka.common.errors.ProducerFencedException;
import java.util.*;
public class SampleProducer {
public static String topic = "topic-4";
public static void main(String[] args) {
Properties configProperties = new Properties();
//configProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "some-client-id");
configProperties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TXN_ID:" + new Random().nextDouble() + new Random().nextInt());
configProperties.put("acks", "all");
configProperties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
configProperties.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
configProperties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
configProperties.put("value.serializer", JsonSerializer.class);
configProperties.put("bootstrap.servers", "192.168.41.132:9090<http://192.168.41.132:9090>");
KafkaProducer<Integer, DataObject>producer = new KafkaProducer<>(configProperties);
System.out.println("Init Transaction");
producer.initTransactions();
try {
System.out.println("transaction initialised going to begin transaction");
producer.beginTransaction();
System.out.println("Transaction started");
ProducerRecord rec = new ProducerRecord(topic, 5, new DataObject(5, "Hello, World!"));
RecordMetadata metadata = (RecordMetadata) producer.send(rec).get();
System.out.println("The offset of the record we just sent is: " + metadata.offset());
metadata = (RecordMetadata) producer.send(rec).get();
System.out.println("The offset of the record we just sent is: " + metadata.offset());
producer.commitTransaction();
System.out.println("Transaction Committed");
}catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e){
// We can't recover from these exceptions, so our only option is to close the producer and exit.
System.out.println("Connection closed but commit failed. We can't recover");
producer.close();
}catch(KafkaException e) {
// For all other exceptions, just abort the transaction and try again.
System.out.println("Abort Transaction");
producer.abortTransaction();
}catch (Exception x){}
producer.close();
System.out.println("Closed");
}
}
</CODE proucer>
These are my producer console logs
<LOGS producer>
0 [main] INFO org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values:
acks = all
batch.size = 16384
bootstrap.servers = [192.168.41.132:9090<http://192.168.41.132:9090>]
buffer.memory = 33554432
client.id<http://client.id> =
compression.type = none
connections.max.idle.ms<http://connections.max.idle.ms> = 540000
enable.idempotence = true
interceptor.classes = null
key.serializer = class org.apache.kafka.common.serialization.IntegerSerializer
linger.ms<http://linger.ms> = 0
max.block.ms<http://max.block.ms> = 60000
max.in<http://max.in>.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.max.age.ms<http://metadata.max.age.ms> = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms<http://metrics.sample.window.ms> = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.max.ms<http://reconnect.backoff.max.ms> = 1000
reconnect.backoff.ms<http://reconnect.backoff.ms> = 50
request.timeout.ms<http://request.timeout.ms> = 30000
retries = 2147483647
retry.backoff.ms<http://retry.backoff.ms> = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin =60000
sasl.kerberos.service.name<http://sasl.kerberos.service.name> = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
transaction.timeout.ms<http://transaction.timeout.ms> = 60000
transactional.id<http://transactional.id> = TXN_ID:0.5031925219291776-156417066
value.serializer = class com.example.transaction.producer.utils.serde.JsonSerializer
261 [main] INFO org.apache.kafka.clients.producer.KafkaProducer - Instantiated a transactional producer.
265 [main] INFO org.apache.kafka.clients.producer.KafkaProducer - Overriding the defaultmax.in<http://defaultmax.in>.flight.requests.per.connection to 1 since idempontence is enabled.
274 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name bufferpool-wait-time
281 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name buffer-exhausted-records
284 [main] DEBUG org.apache.kafka.clients.Metadata - Updated cluster metadata version 1 to Cluster(id = null, nodes = [192.168.41.132:9090<http://192.168.41.132:9090> (id: -1 rack: null)], partitions = [])
297 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name produce-throttle-time
564 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name connections-closed:
564 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name connections-created:
565 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name bytes-sent-received:
565 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name bytes-sent:
565 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name bytes-received:
566 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name select-time:
567 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name io-time:
573 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name batch-size
574 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name compression-rate
574 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name queue-time
575 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name request-time
575 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name records-per-request
577 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name record-retries
577 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name errors
577 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name record-size-max
579 [main] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name batch-split-rate
582 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Sender - Starting Kafka producer I/O thread.
585 [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 0.11.0.0
585 [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : cb8625948210849f
586 [main] DEBUG org.apache.kafka.clients.producer.KafkaProducer - Kafka producer started
Init Transaction
588 [main] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Transition from state UNINITIALIZED to INITIALIZING
588 [main] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] ProducerId set to -1 with epoch -1
594 [main] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Enqueuing transactional request (type=InitProducerIdRequest, transactionalId=TXN_ID:0.5031925219291776-156417066, transactionTimeoutMs=60000)
598 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Enqueuing transactional request (type=FindCoordinatorRequest, coordinatorKey=TXN_ID:0.5031925219291776-156417066,coordinatorType=TRANSACTION)
598 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Enqueuing transactional request (type=InitProducerIdRequest, transactionalId=TXN_ID:0.5031925219291776-156417066, transactionTimeoutMs=60000)
712 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Initiating connection to node -1 at192.168.41.132:9090.
712 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name node--1.bytes-sent
712 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name node--1.bytes-received
712 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name node--1.latency
712 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.network<http://org.apache.kafka.common.network>.Selector - Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 131072, SO_TIMEOUT = 0 to node -1
712 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Completed connection to node -1. Fetching API versions.
712 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Initiating API versions fetch from node -1.
897 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Recorded API versions for node -1: (Produce(0): 0 to 3 [usable: 3], Fetch(1): 0 to 5 [usable: 5], Offsets(2): 0 to 2 [usable: 2], Metadata(3): 0 to 4 [usable: 4], LeaderAndIsr(4): 0 [usable: 0], StopReplica(5): 0 [usable: 0], UpdateMetadata(6): 0 to 3 [usable: 3], ControlledShutdown(7): 1 [usable: 1], OffsetCommit(8): 0 to 3 [usable: 3], OffsetFetch(9): 0 to 3 [usable: 3], FindCoordinator(10): 0 to 1 [usable: 1],JoinGroup(11): 0 to 2 [usable: 2], Heartbeat(12): 0 to 1 [usable: 1], LeaveGroup(13): 0 to 1 [usable: 1], SyncGroup(14): 0 to 1 [usable: 1], DescribeGroups(15): 0 to 1 [usable: 1], ListGroups(16): 0 to 1 [usable: 1], SaslHandshake(17): 0 [usable: 0], ApiVersions(18): 0 to 1 [usable: 1], CreateTopics(19): 0 to 2 [usable: 2], DeleteTopics(20): 0 to 1 [usable: 1], DeleteRecords(21): 0 [usable: 0], InitProducerId(22): 0 [usable: 0], OffsetForLeaderEpoch(23): 0 [usable: 0], AddPartitionsToTxn(24): 0 [usable: 0], AddOffsetsToTxn(25): 0 [usable: 0], EndTxn(26): 0 [usable: 0], WriteTxnMarkers(27): 0 [usable: 0], TxnOffsetCommit(28): 0 [usable: 0], DescribeAcls(29): 0 [usable: 0], CreateAcls(30): 0 [usable: 0], DeleteAcls(31): 0 [usable: 0], DescribeConfigs(32): 0 [usable: 0], AlterConfigs(33): 0 [usable: 0])
898 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Sender - [TransactionalId TXN_ID:0.5031925219291776-156417066] Sending transactional request (type=FindCoordinatorRequest, coordinatorKey=TXN_ID:0.5031925219291776-156417066,coordinatorType=TRANSACTION) to node 192.168.41.132:9090<http://192.168.41.132:9090> (id: -1 rack: null)
901 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Initiating connection to node 1001 at192.168.41.132:9090.
901 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name node-1001.bytes-sent
902 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name node-1001.bytes-received
902 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.metrics.Metrics - Added sensor with name node-1001.latency
903 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.network<http://org.apache.kafka.common.network>.Selector - Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 131072, SO_TIMEOUT = 0 to node 1001
903 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Completed connection to node 1001.Fetching API versions.
903 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Initiating API versions fetch from node 1001.
905 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Recorded API versions for node 1001: (Produce(0): 0 to 3 [usable: 3], Fetch(1): 0 to 5 [usable: 5], Offsets(2): 0 to 2 [usable: 2], Metadata(3): 0 to 4 [usable: 4], LeaderAndIsr(4): 0 [usable: 0], StopReplica(5): 0 [usable: 0], UpdateMetadata(6): 0 to 3 [usable: 3], ControlledShutdown(7): 1 [usable: 1], OffsetCommit(8): 0 to 3 [usable: 3], OffsetFetch(9): 0 to 3 [usable: 3], FindCoordinator(10): 0 to 1 [usable: 1],JoinGroup(11): 0 to 2 [usable: 2], Heartbeat(12): 0 to 1 [usable: 1], LeaveGroup(13): 0 to 1 [usable: 1], SyncGroup(14): 0 to 1 [usable: 1], DescribeGroups(15): 0 to 1 [usable: 1], ListGroups(16): 0 to 1 [usable: 1], SaslHandshake(17): 0 [usable: 0], ApiVersions(18): 0 to 1 [usable: 1], CreateTopics(19): 0 to 2 [usable: 2], DeleteTopics(20): 0 to 1 [usable: 1], DeleteRecords(21): 0 [usable: 0], InitProducerId(22): 0 [usable: 0], OffsetForLeaderEpoch(23): 0 [usable: 0], AddPartitionsToTxn(24): 0 [usable: 0], AddOffsetsToTxn(25): 0 [usable: 0], EndTxn(26): 0 [usable: 0], WriteTxnMarkers(27): 0 [usable: 0], TxnOffsetCommit(28): 0 [usable: 0], DescribeAcls(29): 0 [usable: 0], CreateAcls(30): 0 [usable: 0], DeleteAcls(31): 0 [usable: 0], DescribeConfigs(32): 0 [usable: 0], AlterConfigs(33): 0 [usable: 0])
1009 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Sender - [TransactionalId TXN_ID:0.5031925219291776-156417066] Sending transactional request (type=InitProducerIdRequest, transactionalId=TXN_ID:0.5031925219291776-156417066, transactionTimeoutMs=60000) to node 192.168.41.132:9090<http://192.168.41.132:9090> (id: 1001 rack: null)
9491 [kafka-producer-network-thread | producer-1] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] ProducerId set to 4001 with epoch 0
9491 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Transition from state INITIALIZING to READY
transaction initialised going to begin transaction
9491 [main] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Transition from state READY to IN_TRANSACTION
Transaction started
9491 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient - Sending metadata request (type=MetadataRequest, topics=topic-4) to node -1
9491 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.Metadata - Updated cluster metadata version 2 to Cluster(id = 0WtNXiFvT_C6V9Uo1zPGVQ, nodes = [192.168.41.132:9090<http://192.168.41.132:9090> (id: 1001 rack: null)],partitions = [Partition(topic = topic-4, partition = 0, leader = 1001, replicas = [1001], isr =[1001])])
9523 [main] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Begin adding new partition topic-4-0 to transaction
9523 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Enqueuing transactional request (type=AddPartitionsToTxnRequest, transactionalId=TXN_ID:0.5031925219291776-156417066, producerId=4001, producerEpoch=0, partitions=[topic-4-0])
9523 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Sender - [TransactionalId TXN_ID:0.5031925219291776-156417066] Sending transactional request (type=AddPartitionsToTxnRequest, transactionalId=TXN_ID:0.5031925219291776-156417066, producerId=4001, producerEpoch=0, partitions=[topic-4-0]) to node 192.168.41.132:9090<http://192.168.41.132:9090> (id: 1001 rack: null)
9538 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.TransactionManager - [TransactionalId TXN_ID:0.5031925219291776-156417066] Successfully added partitions [topic-4-0] to transaction