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.

[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!

[stillya]

Hi, @Wzy19930507 and @sobychacko.
I've decided to lend a hand to already an existing pull request and created own with a new interface to enable the management of transactionIdSuffix. Could you please take a look?