GH-2852: support fixed transaction id suffix #2913
Conversation
add properties `maxCache` at `DefaultKafkaProducerFactory`, setting `maxCache` greater than zero can reuse `transactional.id`. Resolves spring-projects#2852
| } | ||
| } | ||
| else { | ||
| suffix = String.valueOf(this.transactionIdSuffix.getAndIncrement()); |
There was a problem hiding this comment.
Nice work, and thanks for your effort! However, I think just incrementing may not be sufficient. The goal of GH-2852 is to allow users to set how transactionIds are issued. For example, with a topic having 9 partitions, 3 replicas of my service (each with 3 producers), and limits on 9 txIds, restarting my service becomes a bit of a pain. It takes time to distribute transaction IDs correctly(without getting Fenced all the time), so instead of that, I would prefer to use external storage as a source of truth.
I suggest introducing a new interface for generating transactionIdSuffix with a default increment implementation. WDYT?
There was a problem hiding this comment.
I suggest introducing a new interface for generating transactionIdSuffix with a default increment implementation. WDYT?
Via new interface, the first instance of service increments transactionIdSuffix from 0 and the second instance of service increments from 3?
If I understand correctly, I like it.
At the same time we need to monitor the generated trasaction.id, such as provide method return current transactionIdSuffix or pushlish event carry curr transactionIdSuffix at DefaultKafkaProducerFactory.destroy().
As an alternative, write a unique ID in the transactionIdPrefix, such as UUD is feasible?
There was a problem hiding this comment.
Via the new interface, does the first instance of the service increment transactionIdSuffix from 0, and the second instance increments it from 3?
It actually depends on the implementation. It could distribute transactionIdSuffix between replicas as described, or it could provide the first free suffix from the source of truth storage. However, the default implementation remains the same as it is now.
At the same time, we need to monitor the generated transaction.id. For example, provide a method to return the current transactionIdSuffix or publish an event carrying the current transactionIdSuffix at DefaultKafkaProducerFactory.destroy().
Certainly, we need this functionality. Therefore, the interface must include a contract to retrieve the transactionId and return it back.
As an alternative, writing a unique ID in the transactionIdPrefix, such as UUID, is feasible?
In my environment, it's not an option because I have a mask for transactionIds like topic-name-{0..n}(e.g. mytopic-1).
This comment was marked as abuse.
This comment was marked as abuse.
Sorry, something went wrong.
Introducing a new interface for generating transactionIdSuffix with a default increment implementation. Introducing a new interface for return the current transactionIdSuffix. Resolves spring-projects#2852
|
Hi @Wzy19930507, I am taking a look at this PR. Could you let me know if you added the new interface that @stillya suggested? I don't see that in the PR yet. |
|
|
||
| Since 3.1, when setting `maxCache` greater than zero can reuse `transactional.id` within a specific range. | ||
| When a transaction producer is requested and `transactional.id` all in use, throw a `NoProducerAvailableException`. | ||
| User can use then use a RetryTemplate configured to retry that exception, with a suitably configured back off. |
There was a problem hiding this comment.
Change the wording to - "We can then use a"
| When setting `maxCache` to 5, `transactional.id` is `my.txid.`++`{0-4}`+. | ||
|
|
||
| IMPORTANT: When use `KafkaTransactionManager` in the `ConcurrentMessageListenerContainer`, `maxCache` must be greater than `concurrency`, also be careful nested transaction. | ||
|
|
There was a problem hiding this comment.
"also be careful about".
It would be beneficial if you could add a few sentences detailing why the maxCache needs to be greater than concurrency. Same thing about nested transactions.
| public final boolean initTransactionIdSuffix(int transactionIdSuffix) { | ||
| Assert.isTrue(transactionIdSuffix >= 0, "'transactionIdSuffix' initial value must greater than or equal 0"); | ||
| return this.transactionIdSuffix.compareAndSet(0, transactionIdSuffix); | ||
| } |
There was a problem hiding this comment.
Hi, @sobychacko new interface is here. DefaultKafkaProducerFactory use only
There was a problem hiding this comment.
Thank you, but I didn't quite mean that, so I've decided to lend a hand. Could you please take a look?
Add properties
maxCacheatDefaultKafkaProducerFactory, settingmaxCachegreater than zero can reusetransactional.id.If maxCache is 3, the suffix corresponding to the first prefix is 0-2, corresponding to the second prefix is 3-5.
Add new interface to initialize transactionIdSuffix.
Resolves #2852