Impossible to set different transaction prefix for consumer- and producer-based transactions

[l0co]

As stated in the docs:

There is a dichotomy when specifying this property in that, when running multiple instances of the application with EOSMode.ALPHA, it must be the same on all instances to satisfy fencing zombies (also mentioned in the overview) when producing records on a listener container thread. However, when producing records using transactions that are not started by a listener container, the prefix has to be different on each instance.

Please, clarify how it can be achievable with Spring Kafka (2.7.9) because we don't see this option. You either can set static or randomized transactional id for both instances with spring.kafka.producer.transaction-id-prefix.

[garyrussell]

Please don't ask questions in issues - did you not read the issue template?

The transaction-id-prefix property sets it on the producer factory and will be used by the KafkaTemplate by default.

You can define a second template bean for producer-initiated transactions and give it a different transaction id prefix.

	/**
	 * Set a transaction id prefix to override the prefix in the producer factory.
	 * @param transactionIdPrefix the prefix.
	 * @since 2.3
	 */
	public void setTransactionIdPrefix(String transactionIdPrefix) {

If your brokers are 2.5 or later you don't have to worry about this at all and always use a randomized value - EOSMode has been BETA (now called V2) since 2.6.

[l0co]

Hi, I'm sorry but I haven't read the template and apparently missed Discussion feature on GitHub recently :)

Thank you for your answer. We don't have an option to migrate to >= 2.5 version, so we are stuck now in lower version. We have tested this approach and it seems there are at least a few problems involved.

The first thing is that when you turn on transactions, the KafkaTemplate has to be used from @Transactional method, and it this scenario the transactional id prefix will anyway be used from the KafkaTransactionManager, not from the KafkaTemplate.

On the other hand - using different KafkaTemplate (or KafkaTransactionManager) for consumer and produced-based transactions seems to lead to an awkward design, because it somehow kills the AOP @Transactional idea. Please look at this example:

@Service class MyService {
  @Transactional
  void sendUserNotitcation() {
    kafkaTemplate.send(...);
  }
}

Designing this (and other) service interfaces we don't know at this moment whether this method will be called from producer- or consumer-based transaction. Having two methods here would lead to cumbersome service interface. And it might not be even possible to be accomplished, considering deeper call chains.

Why not to expose two properties with transactionId prefixes for consumer- and producer-based transactions in Spring Kafka?

[garyrussell]

Why not...

That is a question for spring-boot; not here.

However, it would be difficult for boot because it only auto configures one template (and will configure none if there are other template(s) defined).

You can use similar logic to what we do in spring-integration-kafka - check if there is already a transaction and use that and, if not, start a new one programmatically using executeInTransaction().

You should not use @Transactional in this case.

See https://github.com/spring-projects/spring-integration/blob/a843dae1512bdfb1308ee8b3f7c59754c4ae5a21/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java#L509-L514

and

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

Lines 705 to 716 in 8e7c72f

	/**
	* Return true if the template is currently running in a transaction on the calling
	* thread.
	* @return true if a transaction is running.
	* @since 2.2.1
	*/
	@Override
	public boolean inTransaction() {
	return this.transactional && (this.producers.get() != null
	|| TransactionSynchronizationManager.getResource(this.producerFactory) != null
	|| TransactionSynchronizationManager.isActualTransactionActive());
	}

In your case, though, you can check if the primary template is inTransaction() and, if not, use the secondary template.

You can put this logic in an abstraction so you wouldn't need it everywhere.

[l0co]

We have solved it in a bit different way, but similar idea, overriding DefaultKafkaProducerFactory with createProducer() custom implementation.

However, I've started this discussion because I feel Spring Kafka/Boot knowns what should be done, and even it tells what should be done very precisely in the docs, at the same time not giving a clear way to achieve it, and it's a thing that can be easily improved. I've spent the whole day to "hack" the current behavior, but mostly on looking for the right hook point to use. And, this is behavior, which has to be implemented somehow by anyone who uses transactionally Kafka < 2.5 and I can imagine tons of production systems which are not upgraded to latest Kafka.

Please refer:

1. https://stackoverflow.com/questions/52220225/spring-kafka-and-transactions
2. https://stackoverflow.com/questions/61345851/producerfencedexception-after-kafka-consumer-group-rebalancing

Also not giving the clear answer.

[garyrussell]

I tried to unlock and reopen your original issue (it was locked when I converted it to a discussion), but after unlocking, GitHub wouldn't let me reopen it.

So I opened this instead... #2016

[garyrussell]

I opened a PR against Boot - the functionality is already there; we just need to expose the property.