[BUG] NOT_COORDINATOR Error always happen randomly when enable transactionCoordinator

[fengyongshe]

Describe the bug When enable transactionCoordinator by setting kafkaTransactionCoordinatorEnabled=true , The exception keeps occured in kafka client org.apache.kafka.common.errors.TimeoutException: Timeout expired after 10000 milliseconds while awaiting InitProducerId

And debug msg: [2023-01-17 01:07:14,223] DEBUG [Producer clientId=producer-xgq0e21b, transactionalId=xgq0e21b] Received INIT_PRODUCER_ID response from node 1599868786 for request with header RequestHeader(apiKey=INIT_PRODUCER_ID, apiVersion=1, clientId=producer-xgq0e21b, correlationId=62): InitProducerIdResponseData(throttleTimeMs=0, errorCode=16, producerId=-1, producerEpoch=-1) (org.apache.kafka.clients.NetworkClient)

The errocode=16 means NOT_COORDINATOR

This is caused by inconsistently function to generate the partitionId in the transaction , there are two place to use partition id 1) Get topic broker for TransactionCoordinator(based on __transaction_state-{partition-id}) , the partition id is generated by

 TransactionCoordinator
  public static int partitionFor(String transactionalId, int transactionLogNumPartitions) {
        return MathUtils.signSafeMod(
                transactionalId.hashCode(),
                transactionLogNumPartitions
        );
    }

2) Get TransactionMeta from cache according to the partition id

TransactionStateManager
 public int partitionFor(String transactionalId) {
        return Utils.abs(transactionalId.hashCode()) % transactionTopicPartitionCount;
   }

For Example, when setting props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "qlf8pp81") in kafka client , The partitionId in the two place FOR "qlf8pp81" is different : 16 for getting topic broker for TransactionCoordinator 34 for getting TransactionMetadata from Transaction Cache

The inconsistently function to generate the partitionId will trigger the exception in transactionStateManager

              Map<String, TransactionMetadata> metadataMap = transactionMetadataCache.get(partitionId);
                if (metadataMap == null) {
                    return new ErrorsAndData<>(Errors.NOT_COORDINATOR, Optional.empty());
                }

To Reproduce Steps to reproduce the behavior:

0. Enable transaction by setting ：kafkaTransactionCoordinatorEnabled=true and transactionLogNumPartitions=50

1. Choose the different partition Id generated for same transaction id , eg: qlf8pp81, xgq0e21b。 Ensure that there is no transactionMetadata for the txid in transactionMetadataCache(Map<Integer {partitionId}, Map<tx_id, TransactionMetadata>>) , you can add debug info in TransactionStateManager to print the cache : log.debug("The Cache content before Add Transaction State: {}", transactionMetadataCache); The content in cache for example: {2={}, 22={}, 6={}, 26={}}

2. Kafka Example use transaction , kernel node as below ..... props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "xgq0e21b") ; KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props); producer.initTransactions(); producer.beginTransaction(); Future metadataFuture = producer.send(new ProducerRecord<>("cptest", "world")); .... producer.commitTransaction(); producer.close();

3. Run the test , the exception will occure , add the debug in transactionStateManater

   2023-01-17T01:06:47,850+0800 [pulsar-io-4-16] DEBUG io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionStateManager - The partitionId : 44 for transactionId: xgq0e21b
   2023-01-17T01:06:47,850+0800 [pulsar-io-4-16] DEBUG io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionStateManager - The Cache content before Add Transaction State: {2={}, 22={}, 6={}, 26={}}
   2023-01-17T01:06:47,850+0800 [pulsar-io-4-16] DEBUG io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionStateManager - Transaction Metadata Cache content for partition Id:44 for metadataMap:null
   2023-01-17T01:06:47,850+0800 [pulsar-io-4-16] DEBUG io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionStateManager - There is not coordiantor for partitionId: 44
   2023-01-17T01:06:47,850+0800 [pulsar-io-4-16] DEBUG io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionStateManager - The Content in transactionMetadataCache : {2={}, 22={}, 6={}, 26={}}

   

Expected behavior Use the same function to generate the partition id (Function: partitionFor(String transactionalId) )in TransactionCoordinator and TransactionStateManager . The bug is resolved.

When load bundle to a broker , kop.NamespaceBundleOwnershipListenerImpl will trigger to put the empty TransactionMetadataMap for the transaction state partition belong to the the bundle.

[nextadmin-beep]

i have the same problem