Kafkajs: Unexpected metadata causes IllegalStateException
We ran into an java.lang.IllegalStateException while testing the EoS implementation and we are not sure what is causing it.
The scenario is as follows: in a loop, we create a KafkaJs client and a producer and we write a number of messages in a single transaction, then discard the client.
What we observe is that after a variable number of iterations the following exception happens:
kafka1_1 | [2019-02-28 19:34:07,949] ERROR TransactionMetadata(transactionalId=transactional-id, producerId=382001, producerEpoch=877, txnTimeoutMs=60000, state=Empty, pendingState=Some(Ongoing), topicPartitions=Set(), txnStartTimestamp=1551382447950, txnLastUpdateTimestamp=1551382447936)'s transition to TxnTransitMetadata(producerId=382001, producerEpoch=877, txnTimeoutMs=60000, txnState=Ongoing, topicPartitions=Set(test-topic-1551382421868-1), txnStartTimestamp=1551382447942, txnLastUpdateTimestamp=1551382447942) failed: this should not happen (kafka.coordinator.transaction.TransactionMetadata)
kafka1_1 | [2019-02-28 19:34:07,953] ERROR [KafkaApi-0] Error when handling request {transactional_id=transactional-id,producer_id=382001,producer_epoch=877,topics=[{topic=test-topic-1551382421868,partitions=[1]}]} (kafka.server.KafkaApis)
kafka1_1 | java.lang.IllegalStateException: TransactionalId transactional-id failed transition to state TxnTransitMetadata(producerId=382001, producerEpoch=877, txnTimeoutMs=60000, txnState=Ongoing, topicPartitions=Set(test-topic-1551382421868-1), txnStartTimestamp=1551382447942, txnLastUpdateTimestamp=1551382447942) due to unexpected metadata
(further detail below).
The Kafka cluster remains up, however, a subsequent run of the reproducer results in KafkaJSNumberOfRetriesExceeded after a number of retries on the CONCURRENT_TRANSACTIONS error. This is regardless of if we use the same topic.
Reproducer:
const ip = require("ip");
const kafkajs = require("kafkajs");
const id = Date.now();
const transactionalId = "transactional-id";
const host = process.env.HOST_IP || ip.address();
const broker = `${host}:9092`;
const topic = `test-topic-${id}`;
const acks = -1;
const maxInFlightRequests = 1;
const connectionTimeout = 1000;
const key = `sink-1-${id}`;
async function setupAdmin() {
const client = new kafkajs.Kafka({ clientId: "reproducer", brokers: [broker] });
const admin = client.admin();
try {
await admin.connect();
await admin.createTopics({
waitForLeaders: true,
topics: [{ topic, numPartitions: 4, configEntries: [] }],
});
} finally {
await admin.disconnect();
}
console.log("ADMIN SETUP DONE");
}
function createMessages(num) {
const messages = [];
for (let ii = 0; ii < num; ii++) {
messages.push({ key, value: `${ii}` });
}
return messages;
}
async function runOneProducer(num) {
const messages = createMessages(num);
const client = new kafkajs.Kafka({
clientId: "reproducer",
brokers: [broker],
connectionTimeout,
maxInFlightRequests,
});
const producer = client.producer({
// An idempotent producer enforces EoS messaging
// idempotent: transactional ? true : false,
idempotent: true,
transactionalId,
});
await producer.connect();
const transaction = await producer.transaction();
try {
await transaction.send({
acks,
topic,
messages,
});
} catch (e) {
try {
await transaction.abort();
} catch (e) {
throw e;
}
throw e;
}
await transaction.commit();
await producer.disconnect();
}
async function runProducerReplicator() {
await setupAdmin();
const repeats = 5000;
const numMessages = 128;
for (let ii = 0; ii < repeats; ii++) {
await runOneProducer(numMessages);
if (ii % 20 === 0) {
console.log(ii);
}
}
}
runProducerReplicator();
Kafka Log:
kafka1_1 | [2019-02-28 19:34:07,944] INFO [TransactionCoordinator id=0] Initialized transactionalId transactional-id with producerId 382001 and producer epoch 876 on partition __transaction_state-31 (kafka.coordinator.transaction.TransactionCoordinator)
kafka1_1 | [2019-02-28 19:34:07,936] INFO [TransactionCoordinator id=0] Initialized transactionalId transactional-id with producerId 382001 and producer epoch 877 on partition __transaction_state-31 (kafka.coordinator.transaction.TransactionCoordinator)
kafka1_1 | [2019-02-28 19:34:07,949] ERROR TransactionMetadata(transactionalId=transactional-id, producerId=382001, producerEpoch=877, txnTimeoutMs=60000, state=Empty, pendingState=Some(Ongoing), topicPartitions=Set(), txnStartTimestamp=1551382447950, txnLastUpdateTimestamp=1551382447936)'s transition to TxnTransitMetadata(producerId=382001, producerEpoch=877, txnTimeoutMs=60000, txnState=Ongoing, topicPartitions=Set(test-topic-1551382421868-1), txnStartTimestamp=1551382447942, txnLastUpdateTimestamp=1551382447942) failed: this should not happen (kafka.coordinator.transaction.TransactionMetadata)
kafka1_1 | [2019-02-28 19:34:07,953] ERROR [KafkaApi-0] Error when handling request {transactional_id=transactional-id,producer_id=382001,producer_epoch=877,topics=[{topic=test-topic-1551382421868,partitions=[1]}]} (kafka.server.KafkaApis)
kafka1_1 | java.lang.IllegalStateException: TransactionalId transactional-id failed transition to state TxnTransitMetadata(producerId=382001, producerEpoch=877, txnTimeoutMs=60000, txnState=Ongoing, topicPartitions=Set(test-topic-1551382421868-1), txnStartTimestamp=1551382447942, txnLastUpdateTimestamp=1551382447942) due to unexpected metadata
kafka1_1 | at kafka.coordinator.transaction.TransactionMetadata.throwStateTransitionFailure(TransactionMetadata.scala:390)
kafka1_1 | at kafka.coordinator.transaction.TransactionMetadata.completeTransitionTo(TransactionMetadata.scala:326)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$kafka$coordinator$transaction$TransactionStateManager$$updateCacheCallback$1$1.apply$mcV$sp(TransactionStateManager.scala:534)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$kafka$coordinator$transaction$TransactionStateManager$$updateCacheCallback$1$1.apply(TransactionStateManager.scala:526)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$kafka$coordinator$transaction$TransactionStateManager$$updateCacheCallback$1$1.apply(TransactionStateManager.scala:526)
kafka1_1 | at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:250)
kafka1_1 | at kafka.coordinator.transaction.TransactionMetadata.inLock(TransactionMetadata.scala:172)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager.kafka$coordinator$transaction$TransactionStateManager$$updateCacheCallback$1(TransactionStateManager.scala:525)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$appendTransactionToLog$1$$anonfun$apply$mcV$sp$11.apply(TransactionStateManager.scala:620)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$appendTransactionToLog$1$$anonfun$apply$mcV$sp$11.apply(TransactionStateManager.scala:620)
kafka1_1 | at kafka.server.DelayedProduce.onComplete(DelayedProduce.scala:129)
kafka1_1 | at kafka.server.DelayedOperation.forceComplete(DelayedOperation.scala:70)
kafka1_1 | at kafka.server.DelayedProduce.tryComplete(DelayedProduce.scala:111)
kafka1_1 | at kafka.server.DelayedOperationPurgatory.tryCompleteElseWatch(DelayedOperation.scala:232)
kafka1_1 | at kafka.server.ReplicaManager.appendRecords(ReplicaManager.scala:488)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$appendTransactionToLog$1.apply$mcV$sp(TransactionStateManager.scala:614)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$appendTransactionToLog$1.apply(TransactionStateManager.scala:591)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager$$anonfun$appendTransactionToLog$1.apply(TransactionStateManager.scala:591)
kafka1_1 | at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:250)
kafka1_1 | at kafka.utils.CoreUtils$.inReadLock(CoreUtils.scala:256)
kafka1_1 | at kafka.coordinator.transaction.TransactionStateManager.appendTransactionToLog(TransactionStateManager.scala:585)
kafka1_1 | at kafka.coordinator.transaction.TransactionCoordinator.handleAddPartitionsToTransaction(TransactionCoordinator.scala:272)
kafka1_1 | at kafka.server.KafkaApis.handleAddPartitionToTxnRequest(KafkaApis.scala:1744)
kafka1_1 | at kafka.server.KafkaApis.handle(KafkaApis.scala:128)
kafka1_1 | at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69)
kafka1_1 | at java.lang.Thread.run(Thread.java:748)
kafka2_1 | waiting for kafka to be ready
kafka3_1 | waiting for kafka to be ready
KafkaJs Log:
ADMIN SETUP DONE
0
{"level":"ERROR","timestamp":"2019-02-28T19:33:42.609Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":5,"size":20}
20
…
180
{"level":"ERROR","timestamp":"2019-02-28T19:33:48.652Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":4,"size":20}
200
…
860
{"level":"ERROR","timestamp":"2019-02-28T19:34:07.963Z","logger":"kafkajs","message":"[Connection] Response AddPartitionsToTxn(key: 24, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The server experienced an unexpected error when processing the request","correlationId":7,"size":48}
{"level":"ERROR","timestamp":"2019-02-28T19:34:07.964Z","logger":"kafkajs","message":"[Producer] The server experienced an unexpected error when processing the request","retryCount":0,"retryTime":342}
{"level":"ERROR","timestamp":"2019-02-28T19:34:07.971Z","logger":"kafkajs","message":"[Connection] Response EndTxn(key: 26, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":8,"size":10}
(node:14623) UnhandledPromiseRejectionWarning: Unhandled promise rejection (rejection id: 884): KafkaJSProtocolError: The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing
(node:14623) [DEP0018] DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.
KafkaJs Log after second run of reproducer:
ADMIN SETUP DONE
{"level":"ERROR","timestamp":"2019-02-28T19:35:18.152Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":4,"size":20}
{"level":"ERROR","timestamp":"2019-02-28T19:35:18.525Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":5,"size":20}
{"level":"ERROR","timestamp":"2019-02-28T19:35:19.148Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":7,"size":20}
{"level":"ERROR","timestamp":"2019-02-28T19:35:20.358Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":8,"size":20}
{"level":"ERROR","timestamp":"2019-02-28T19:35:22.333Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":9,"size":20}
{"level":"ERROR","timestamp":"2019-02-28T19:35:26.489Z","logger":"kafkajs","message":"[Connection] Response InitProducerId(key: 22, version: 0)","broker":"10.51.16.56:9092","clientId":"reproducer","error":"The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing","correlationId":11,"size":20}
(node:14652) UnhandledPromiseRejectionWarning: Unhandled promise rejection (rejection id: 9): KafkaJSNumberOfRetriesExceeded: The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing
(node:14652) [DEP0018] DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.
plameniv
All 8 comments
The following comment in KAFKA-7088 mentions the same IllegalStateException:
plameniv
on 28 Feb 2019
That's interesting, so you are properly committing or aborting but discarding the client after one commit. I will take a look, thanks.
tulios
on 1 Mar 2019
Running this locally and looking at the logs, this seems to be why the transaction state transition fails:
# Old state
TransactionMetadata(transactionalId=transactional-id, producerId=1, producerEpoch=1804, txnTimeoutMs=60000, state=Empty, pendingState=Some(Ongoing), topicPartitions=Set(), txnStartTimestamp=1551689126534, txnLastUpdateTimestamp=1551689126521)
#New state
TxnTransitMetadata(producerId=1, producerEpoch=1804, txnTimeoutMs=60000, txnState=Ongoing, topicPartitions=Set(test-topic-1551689052613-3), txnStartTimestamp=1551689126527, txnLastUpdateTimestamp=1551689126527)
Not sure how it's possible, but it looks like the current state has a txnStartTimestamp that's higher than the txnStartTimestamp of the state that it's trying to transition to.
Nevon
on 4 Mar 2019
Reading the Kafka code, I'm confused as to how this works. If I'm not mistaken, the timestamps are all set on the broker side:
If the request is handled by a different broker, or if the system clock changes, I would expect this to fail, as the clocks won't be synchronized. They even removed one such check here: https://github.com/apache/kafka/pull/3286/files
Looking at the logs from the reporter, there is something curious:
kafka1_1 | [2019-02-28 19:34:07,944] INFO [TransactionCoordinator id=0] Initialized transactionalId transactional-id with producerId 382001 and producer epoch 876 on partition __transaction_state-31 (kafka.coordinator.transaction.TransactionCoordinator)
kafka1_1 | [2019-02-28 19:34:07,936] INFO [TransactionCoordinator id=0] Initialized transactionalId transactional-id with producerId 382001 and producer epoch 877 on partition __transaction_state-31 (kafka.coordinator.transaction.TransactionCoordinator)
Note how epoch 877 is initialized at 2019-02-28 19:34:07,936 and 876 at 2019-02-28 19:34:07,944. This means that at least according to the logs, epoch 876 was initialized 8ms after 877. If we then look at the difference in txStartTimestamp between the previous state and the one that is being transitioned to, the previous state is 8ms newer than the one that's being transitioned to.
Now, maybe this is a coincidence or I'm misinterpreting the logs, but this looks fishy to me.
Nevon
on 4 Mar 2019
Just a note that I learned about recently, which may explain why we're seeing rather many CONCURRENT_TRANSACTIONS errors.
When we commit a transaction, the coordinator just writes a PrepareCommit message to the log, and then later it asynchronously writes the CompleteCommit message. This means that if you initialize another transaction before everything has been written, you'll get a CONCURRENT_TRANSACTIONS error.
I created #304, as it's related.
I've also been trying to reproduce this in different scenarios.
- [x] Reproduces in Kafka 1.1
- [x] Reproduces in single-broker cluster (1.1 or 0.11)
- [ ] Does not reproduce with Java client
This certainly indicates that it is an issue in KafkaJS, but my suspicion is that it's timing-related, and running my Java consumer is A LOT slower than KafkaJS. So either I'm doing something weird, or there's something going on in the Java client that's waiting for something.
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.clients.producer.*;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import java.util.Properties;
import java.util.concurrent.Future;
public class KafkaTest {
@Test
@DisplayName("Idempotent producer 0.11.0.2")
public void producer() throws Exception {
long noOfProducers = 5000;
long noOfMessages = 128;
String topic = "test-topic";
Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.3.220.56:9092");
producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "transactional-producer");
producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-transactional-id");
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
for (int producers = 0; producers < noOfProducers; producers++) {
Producer<String, String> producer = new KafkaProducer<>(producerConfig);
producer.initTransactions(); // initiate transactions
try {
producer.beginTransaction(); // begin transactions
for (int i = 0; i < noOfMessages; i++) {
Future<RecordMetadata> result = producer
.send(new ProducerRecord<String, String>(topic, "key-" + i, "message-" + i));
}
producer.commitTransaction();
} catch (KafkaException e) {
// For all other exceptions, just abort the transaction and try again.
producer.abortTransaction();
}
producer.close();
}
}
}
@plameniv we ran several tests using your reproducer, and the results are inconclusive. When running against a local Kafka (non-dockerized), it works all the time. We can only reproduce the problem when running on a dockerized Kafka, and like @Nevon said it is related to a timing issue. We can't reproduce on the Java client because it can't go through the same flow fast enough. I've added a sleep call to your reproducer (right after disconnect), and managed to get it to work all the time. Kafka is removing this check, and it will make commit a blocking call, which should fix this for good.
Do you have a use case where you have to trash the producer all the time? When using transactions you should keep the same producer, so I'm assuming you are just testing the edge cases. I don't think this is a blocker for the 1.5.0 release, but I'll wait for your answer.
Thanks again for the great issues.
tulios
on 5 Mar 2019
agreed, this is an edge case and not a blocker for the 1.5.0 release
sklose
on 5 Mar 2019
I will release 1.5.0 then, @sklose @plameniv thanks for shaping up this release!
tulios
on 5 Mar 2019
Related issues
pimpelsang
·
3Comments
paambaati
·
4Comments
Brazier85
·
4Comments
esuarezz
·
6Comments
oemergenc
·
5Comments