{"action":"opened","issue":{"url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/660","repository_url":"https://api.github.com/repos/spring-projects/spring-kafka","labels_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/660/labels{/name}","comments_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/660/comments","events_url":"https://api.github.com/repos/spring-projects/spring-kafka/issues/660/events","html_url":"https://github.com/spring-projects/spring-kafka/issues/660","id":316841998,"number":660,"title":"The producer attempted to use a producer id which is not currently assigned to its transactional id","user":{"login":"pedrohh","id":6523062,"avatar_url":"https://avatars0.githubusercontent.com/u/6523062?v=4","gravatar_id":"","url":"https://api.github.com/users/pedrohh","html_url":"https://github.com/pedrohh","followers_url":"https://api.github.com/users/pedrohh/followers","following_url":"https://api.github.com/users/pedrohh/following{/other_user}","gists_url":"https://api.github.com/users/pedrohh/gists{/gist_id}","starred_url":"https://api.github.com/users/pedrohh/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pedrohh/subscriptions","organizations_url":"https://api.github.com/users/pedrohh/orgs","repos_url":"https://api.github.com/users/pedrohh/repos","events_url":"https://api.github.com/users/pedrohh/events{/privacy}","received_events_url":"https://api.github.com/users/pedrohh/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2018-04-23T14:38:22Z","updated_at":"2018-04-23T14:38:22Z","closed_at":null,"author_association":"NONE","body":"Hello,\r\n\r\nI'm adding transaction support for my application. I'm using the chained transaction manager mentioned [here](https://github.com/spring-projects/spring-kafka/issues/433#issuecomment-355604507), but for simplicity I tested this with only the KafkaTransactionManager (application is written in kotlin and kafka version is 1.0.1).\r\n\r\n1) Configuration\r\n```\r\n@Configuration\r\n@EnableKafka\r\n@EnableTransactionManagement\r\nclass KafkaProducerConfig {\r\n\r\n    @Autowired\r\n    private lateinit var kafkaConfig: KafkaConfig\r\n\r\n    @Bean\r\n    fun producerFactory(): ProducerFactory<String, Entity> {\r\n        val configProps = HashMap<String, Any>()\r\n\r\n        configProps[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = kafkaConfig.bootstrap\r\n        configProps[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java\r\n        configProps[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = JsonSerializer::class.java\r\n\r\n        val factory = DefaultKafkaProducerFactory<String, Entity> (configProps)\r\n        factory.setTransactionIdPrefix(\"txId\")\r\n\r\n        return factory\r\n    }\r\n\r\n    @Bean\r\n    fun kafkaTemplate(): KafkaTemplate<String, Entity> {\r\n        return KafkaTemplate(producerFactory())\r\n    }\r\n\r\n    @Bean\r\n    fun transactionManager(): KafkaTransactionManager<*, *> {\r\n\r\n        return KafkaTransactionManager(producerFactory())\r\n    }\r\n}\r\n```\r\n\r\n2) Implementation\r\n\r\n```\r\n    @Autowired\r\n    private lateinit var kafka: KafkaTemplate<String, Entity>\r\n\r\n    @Transactional\r\n    @RequestMapping(value = [\"/user\"], method = [RequestMethod.POST])\r\n    fun user(@RequestBody user: User): User {\r\n\r\n        kafka.send(\"user_updated\", user)\r\n\r\n        return user\r\n    }\r\n```\r\n\r\nI'm using docker to run an instance of zookeper and another for the kafka broker, and I tried to simulate a container crash (I killed and restart the kafka docker. We can see the exact time the connection was broken in logs bellow -> \"Connection to node 1 could not be established. Broker may not be available\"), and after the kafka service is started I have to call the above endpoint multiple times until the producer exits the error state (exception received is \"The producer attempted to use a producer id which is not currently assigned to its transactional id\"). I even waited more than 10 minutes before calling the endpoint to see if the producer recovered from it, but it won't work until I call that endpoint 2/3 times.\r\n\r\nIs there any configuration that I'm missing?\r\n\r\nThanks,\r\nPedro\r\n\r\n> 2018-04-23 15:17:01.548  INFO 2239 --- [nio-8010-exec-1] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: \r\n> \tacks = 1\r\n> \tbatch.size = 16384\r\n> \tbootstrap.servers = [http://localhost:9092]\r\n> \tbuffer.memory = 33554432\r\n> \tclient.id = \r\n> \tcompression.type = none\r\n> \tconnections.max.idle.ms = 540000\r\n> \tenable.idempotence = true\r\n> \tinterceptor.classes = null\r\n> \tkey.serializer = class org.apache.kafka.common.serialization.StringSerializer\r\n> \tlinger.ms = 0\r\n> \tmax.block.ms = 60000\r\n> \tmax.in.flight.requests.per.connection = 5\r\n> \tmax.request.size = 1048576\r\n> \tmetadata.max.age.ms = 300000\r\n> \tmetric.reporters = []\r\n> \tmetrics.num.samples = 2\r\n> \tmetrics.recording.level = INFO\r\n> \tmetrics.sample.window.ms = 30000\r\n> \tpartitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner\r\n> \treceive.buffer.bytes = 32768\r\n> \treconnect.backoff.max.ms = 1000\r\n> \treconnect.backoff.ms = 50\r\n> \trequest.timeout.ms = 30000\r\n> \tretries = 0\r\n> \tretry.backoff.ms = 100\r\n> \tsasl.jaas.config = null\r\n> \tsasl.kerberos.kinit.cmd = /usr/bin/kinit\r\n> \tsasl.kerberos.min.time.before.relogin = 60000\r\n> \tsasl.kerberos.service.name = null\r\n> \tsasl.kerberos.ticket.renew.jitter = 0.05\r\n> \tsasl.kerberos.ticket.renew.window.factor = 0.8\r\n> \tsasl.mechanism = GSSAPI\r\n> \tsecurity.protocol = PLAINTEXT\r\n> \tsend.buffer.bytes = 131072\r\n> \tssl.cipher.suites = null\r\n> \tssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]\r\n> \tssl.endpoint.identification.algorithm = null\r\n> \tssl.key.password = null\r\n> \tssl.keymanager.algorithm = SunX509\r\n> \tssl.keystore.location = null\r\n> \tssl.keystore.password = null\r\n> \tssl.keystore.type = JKS\r\n> \tssl.protocol = TLS\r\n> \tssl.provider = null\r\n> \tssl.secure.random.implementation = null\r\n> \tssl.trustmanager.algorithm = PKIX\r\n> \tssl.truststore.location = null\r\n> \tssl.truststore.password = null\r\n> \tssl.truststore.type = JKS\r\n> \ttransaction.timeout.ms = 60000\r\n> \ttransactional.id = txId0\r\n> \tvalue.serializer = class org.springframework.kafka.support.serializer.JsonSerializer\r\n> \r\n> 2018-04-23 15:17:01.589  INFO 2239 --- [nio-8010-exec-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-1, transactionalId=txId0] Instantiated a transactional producer.\r\n> 2018-04-23 15:17:01.589  INFO 2239 --- [nio-8010-exec-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-1, transactionalId=txId0] Overriding the default retries config to the recommended value of 2147483647 since the idempotent producer is enabled.\r\n> 2018-04-23 15:17:01.589  INFO 2239 --- [nio-8010-exec-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-1, transactionalId=txId0] Overriding the default acks to all since idempotence is enabled.\r\n> 2018-04-23 15:17:01.642  INFO 2239 --- [nio-8010-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.0.1\r\n> 2018-04-23 15:17:01.642  INFO 2239 --- [nio-8010-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : c0518aa65f25317e\r\n> 2018-04-23 15:17:01.645  INFO 2239 --- [nio-8010-exec-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1, transactionalId=txId0] ProducerId set to -1 with epoch -1\r\n> 2018-04-23 15:17:01.959  INFO 2239 --- [ad | producer-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1, transactionalId=txId0] ProducerId set to 18001 with epoch 0\r\n> 2018-04-23 15:17:15.032  WARN 2239 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1, transactionalId=txId0] Connection to node 1 could not be established. Broker may not be available.\r\n> 2018-04-23 15:17:19.151  WARN 2239 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1, transactionalId=txId0] Error while fetching metadata with correlation id 25 : {user_updated=INVALID_REPLICATION_FACTOR}\r\n> 2018-04-23 15:17:19.293  WARN 2239 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1, transactionalId=txId0] Error while fetching metadata with correlation id 26 : {user_updated=LEADER_NOT_AVAILABLE}\r\n> 2018-04-23 15:17:19.402  WARN 2239 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1, transactionalId=txId0] Error while fetching metadata with correlation id 27 : {user_updated=LEADER_NOT_AVAILABLE}\r\n> 2018-04-23 15:17:19.514  WARN 2239 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1, transactionalId=txId0] Error while fetching metadata with correlation id 28 : {user_updated=LEADER_NOT_AVAILABLE}\r\n> 2018-04-23 15:17:19.625  WARN 2239 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1, transactionalId=txId0] Error while fetching metadata with correlation id 29 : {user_updated=LEADER_NOT_AVAILABLE}\r\n> 2018-04-23 15:17:19.733  WARN 2239 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1, transactionalId=txId0] Error while fetching metadata with correlation id 30 : {user_updated=LEADER_NOT_AVAILABLE}\r\n> 2018-04-23 15:17:36.906 ERROR 2239 --- [ad | producer-1] o.a.k.clients.producer.internals.Sender  : [Producer clientId=producer-1, transactionalId=txId0] Aborting producer batches due to fatal error\r\n> \r\n> org.apache.kafka.common.KafkaException: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \tat org.apache.kafka.clients.producer.internals.TransactionManager$AddPartitionsToTxnHandler.handleResponse(TransactionManager.java:1037) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:905) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:101) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:482) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:474) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216) [kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163) [kafka-clients-1.0.1.jar:na]\r\n> \tat java.lang.Thread.run(Thread.java:748) [na:1.8.0_152]\r\n> Caused by: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:36.908 ERROR 2239 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='eu.spacehatch.api.entities.model.Entity@51dc41f1' to topic user_updated:\r\n> \r\n> org.apache.kafka.common.KafkaException: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \tat org.apache.kafka.clients.producer.internals.TransactionManager$AddPartitionsToTxnHandler.handleResponse(TransactionManager.java:1037) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:905) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:101) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:482) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:474) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216) [kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163) [kafka-clients-1.0.1.jar:na]\r\n> \tat java.lang.Thread.run(Thread.java:748) [na:1.8.0_152]\r\n> Caused by: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:37.014 ERROR 2239 --- [nio-8010-exec-5] o.s.k.t.KafkaTransactionManager          : Commit exception overridden by rollback exception\r\n> \r\n> org.apache.kafka.common.KafkaException: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \tat org.apache.kafka.clients.producer.internals.TransactionManager$AddPartitionsToTxnHandler.handleResponse(TransactionManager.java:1037) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:905) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:101) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:482) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:474) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163) ~[kafka-clients-1.0.1.jar:na]\r\n> \tat java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_152]\r\n> Caused by: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:37.025 ERROR 2239 --- [nio-8010-exec-5] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state] with root cause\r\n> \r\n> org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:44.730 ERROR 2239 --- [nio-8010-exec-7] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.transaction.CannotCreateTransactionException: Could not create Kafka transaction; nested exception is org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state] with root cause\r\n> \r\n> org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:52.725 ERROR 2239 --- [nio-8010-exec-9] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.transaction.CannotCreateTransactionException: Could not create Kafka transaction; nested exception is org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state] with root cause\r\n> \r\n> org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:53.618 ERROR 2239 --- [nio-8010-exec-2] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.transaction.CannotCreateTransactionException: Could not create Kafka transaction; nested exception is org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state] with root cause\r\n> \r\n> org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:55.980 ERROR 2239 --- [nio-8010-exec-3] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.transaction.CannotCreateTransactionException: Could not create Kafka transaction; nested exception is org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state] with root cause\r\n> \r\n> org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:56.909 ERROR 2239 --- [nio-8010-exec-4] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.transaction.CannotCreateTransactionException: Could not create Kafka transaction; nested exception is org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state] with root cause\r\n> \r\n> org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id\r\n> \r\n> 2018-04-23 15:17:57.768  INFO 2239 --- [nio-8010-exec-7] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: \r\n> \tacks = 1\r\n> \tbatch.size = 16384\r\n> \tbootstrap.servers = [http://localhost:9092]\r\n> \tbuffer.memory = 33554432\r\n> \tclient.id = \r\n> \tcompression.type = none\r\n> \tconnections.max.idle.ms = 540000\r\n> \tenable.idempotence = true\r\n> \tinterceptor.classes = null\r\n> \tkey.serializer = class org.apache.kafka.common.serialization.StringSerializer\r\n> \tlinger.ms = 0\r\n> \tmax.block.ms = 60000\r\n> \tmax.in.flight.requests.per.connection = 5\r\n> \tmax.request.size = 1048576\r\n> \tmetadata.max.age.ms = 300000\r\n> \tmetric.reporters = []\r\n> \tmetrics.num.samples = 2\r\n> \tmetrics.recording.level = INFO\r\n> \tmetrics.sample.window.ms = 30000\r\n> \tpartitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner\r\n> \treceive.buffer.bytes = 32768\r\n> \treconnect.backoff.max.ms = 1000\r\n> \treconnect.backoff.ms = 50\r\n> \trequest.timeout.ms = 30000\r\n> \tretries = 0\r\n> \tretry.backoff.ms = 100\r\n> \tsasl.jaas.config = null\r\n> \tsasl.kerberos.kinit.cmd = /usr/bin/kinit\r\n> \tsasl.kerberos.min.time.before.relogin = 60000\r\n> \tsasl.kerberos.service.name = null\r\n> \tsasl.kerberos.ticket.renew.jitter = 0.05\r\n> \tsasl.kerberos.ticket.renew.window.factor = 0.8\r\n> \tsasl.mechanism = GSSAPI\r\n> \tsecurity.protocol = PLAINTEXT\r\n> \tsend.buffer.bytes = 131072\r\n> \tssl.cipher.suites = null\r\n> \tssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]\r\n> \tssl.endpoint.identification.algorithm = null\r\n> \tssl.key.password = null\r\n> \tssl.keymanager.algorithm = SunX509\r\n> \tssl.keystore.location = null\r\n> \tssl.keystore.password = null\r\n> \tssl.keystore.type = JKS\r\n> \tssl.protocol = TLS\r\n> \tssl.provider = null\r\n> \tssl.secure.random.implementation = null\r\n> \tssl.trustmanager.algorithm = PKIX\r\n> \tssl.truststore.location = null\r\n> \tssl.truststore.password = null\r\n> \tssl.truststore.type = JKS\r\n> \ttransaction.timeout.ms = 60000\r\n> \ttransactional.id = txId1\r\n> \tvalue.serializer = class org.springframework.kafka.support.serializer.JsonSerializer\r\n> \r\n> 2018-04-23 15:17:57.772  INFO 2239 --- [nio-8010-exec-7] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-2, transactionalId=txId1] Instantiated a transactional producer.\r\n> 2018-04-23 15:17:57.772  INFO 2239 --- [nio-8010-exec-7] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-2, transactionalId=txId1] Overriding the default retries config to the recommended value of 2147483647 since the idempotent producer is enabled.\r\n> 2018-04-23 15:17:57.772  INFO 2239 --- [nio-8010-exec-7] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-2, transactionalId=txId1] Overriding the default acks to all since idempotence is enabled.\r\n> 2018-04-23 15:17:57.780  INFO 2239 --- [nio-8010-exec-7] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.0.1\r\n> 2018-04-23 15:17:57.780  INFO 2239 --- [nio-8010-exec-7] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : c0518aa65f25317e\r\n> 2018-04-23 15:17:57.783  INFO 2239 --- [nio-8010-exec-7] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-2, transactionalId=txId1] ProducerId set to -1 with epoch -1\r\n> 2018-04-23 15:17:58.387  INFO 2239 --- [ad | producer-2] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-2, transactionalId=txId1] ProducerId set to 19000 with epoch 0\r\n\r\n"}}
{"actor":{"display_login":"pedrohh"}}