Provide a way to customize a transactionIdSuffix

[stillya]

Expected Behavior

Provide strategy to customize a transactionIdSuffix.

Current Behavior

Currently, spring-kafka use AtomicInteger for transactionIdSuffix.

Context

In my environment, I work with restricted transactionId within a specific range(e.g. topic-{1-5}).
Currently, spring-kafka use AtomicInteger for transactionIdSuffix, it would be beneficial to have the flexibility to implement a custom strategy for transactionIdSuffix.
This could involve actions such as resetting the counter after reaching a limit or a storage for managing transactionsIds.

I'm happy to contribute a PR once I know what your opinion on this feature is.

[garyrussell]

This makes sense, although you can't reset it if there is still a producer with that prefix; the reset would have to eject the existing producer from the cache.

A PR would certainly be welcome; thanks!

[Wzy19930507]

Restricted transactionId within a specific range(e.g. topic-{1-5})

If all transactionId are in use, DefaultKafkaProducerFactory.createTransactionalProducer's behavior needs to change.
such as return null, wait some time or threw exception.

Is it reasonable to implement a transaction connection pool?

[stillya]

Is it reasonable to implement a transaction connection pool?

I don't think it's feasible because a transaction is a shared resource among many instances rather than a cross-one, so we can't suggest which transactionId is being used. In my opinion, the default strategy could be a round-robin approach. When we reach the limit, the counter resets. After each FencedException we got on producing, we increment the counter and eject existing producer from the cache.

[Wzy19930507]

Hi, @stillya

shared resource among many instances rather than a cross-one, so we can't suggest which transactionId is being used

I'm sorry, I didn't get your point.

---

When we reach the limit, the counter resets. After each FencedException we got on producing, we increment the counter and eject existing producer from the cache.

This scene is similar to Executors.newFixedThreadPool

• If tx-producer exists in the cache, return from cache
• if tx-producer not exists in the cache, check whether the alive tx-producer have reached the limit
  • if not reached the limit, create tx-producer with unused txid in range {1-5}
  • if reached the limit, execute reject policy
• tx-producer have keepAliveTime
• tx done
  • If tx normally close, return tx-producer to cache
  • If tx abnormal close because of FencedException , close tx-producer

---

Now the strategy of the DefaultKafkaProducerFactory.createTransactionalProducer is similar to Executors.newCachedThreadPool.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

Lines 836 to 854 in c8e0c33

	protected Producer<K, V> createTransactionalProducer(String txIdPrefix) {
	BlockingQueue<CloseSafeProducer<K, V>> queue = getCache(txIdPrefix);
	Assert.notNull(queue, () -> "No cache found for " + txIdPrefix);
	CloseSafeProducer<K, V> cachedProducer = queue.poll();
	while (cachedProducer != null) {
	if (expire(cachedProducer)) {
	cachedProducer = queue.poll();
	}
	else {
	break;
	}
	}
	if (cachedProducer == null) {
	return doCreateTxProducer(txIdPrefix, "" + this.transactionIdSuffix.getAndIncrement(), this::cacheReturner);
	}
	else {
	return cachedProducer;
	}
	}

• If tx-producer exists in the cache, return from cache
• if tx-producer not exists in cache, create tx-producer
• tx-producer have keepAliveTime
• tx done
  • If tx normally close, return tx-producer to cache
  • If tx abnormal close because of FencedException , close tx-producer

[garyrussell]

Probably the simplest solution would be a new property maxCache. When non-zero, we pre-populate a blocking queue with the "available" suffixes.

When a producer is not returned to the cache (either because it was fenced, or because the maxAge expired in expire()) we add that suffix back to the queue for reuse.

When a producer is requested and there are no cached producers, and there are no "available" suffixes, we throw a NoProducerAvailableException.

The user can then use a RetryTemplate configured to retry that exception, with a suitably configured back off.

[Wzy19930507]

Hi, @stillya @garyrussell , If you don't mind, may i pick it up?
I think maxCache is feasible.

[garyrussell]

Sure; thanks!