Polish KafkaMessageListenerContainer

[Wzy19930507]

• remove checkAckMode because ContainerProperties.ackCount and ContainerProperties.ackTime have init value.
• change ListenerConsumer.offsets type to Map<TopicPartition, Long>.
• replace ListenerConsumer.setupSubBatchPerPartition to ContainerProperties.isSubBatchPerPartition.
• add method ListenerConsumer.commitOffsets and ListenerConsumer.commitOffsetsInTransactions
• remove properties ListenerConsumer.this.commitRecovered.
• refactor method ListenerConsumer.processCommits.
• refactor method ListenerConsumer.ackCurrent.

[sobychacko]

@Wzy19930507 There are test failures in the PR: https://github.com/spring-projects/spring-kafka/actions/runs/8115263868/job/22182761295?pr=3090. Could you take a look?

[Wzy19930507]

Introduced a unit test bug from #2287, sorry for it.

When AckMode.RECORD and early is false, the last execute method is RecordInterceptor.afterRecord, but i forget a CountDownLatch to wait RecordInterceptor.afterRecord completes execution.

[sobychacko]

Where did this code (updatePendingOffsets) move to? Do we not need this any longer? I see that you removed the method, but where is this logic now?

[Wzy19930507]

updatePendingOffsets() feel like duplicate in processAck()

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Lines 2024 to 2034 in 5beb8fe

	if (this.isManualImmediateAck) {
	try {
	ackImmediate(cRecord);
	}
	catch (@SuppressWarnings(UNUSED) WakeupException e) {
	// ignore - not polling
	}
	}
	else {
	addOffset(cRecord);
	}

[sobychacko]

But processAck() is not called from the hierarchy of processCommits where updatePendingOffsets() was originally called. I see that there are some duplicates, but still wondering whether it changes the logic a little bit?

[sobychacko]

@Wzy19930507 Any updates on this? This is the only pending item that needs an explanation before merging the PR.

[Wzy19930507]

Sorry for late.

I see that there are some duplicates, but still wondering whether it changes the logic a little bit?

handleAcks has three scenarios and include updatePendingOffsets logic:

1. !Thread.currentThread().equals(this.consumerThread) --> keep isManualImmediateAck thread safe.
2. isManualImmediateAck --> call ackImmediate
3. not isManualImmediateAck --> call addOffset

updatePendingOffsets logic same as scenarios three and see:

	if (!this.isManualImmediateAck) {
		if (!this.isManualAck) {
			updatePendingOffsets();
		}

		private void updatePendingOffsets() {
			ConsumerRecord<K, V> cRecord = this.acks.poll();
			while (cRecord != null) {
				addOffset(cRecord);
				cRecord = this.acks.poll();
			}
		}

[sobychacko]

Thanks for the clarification. One last question on this. In the processCommits() method, we have this logic:

if (!this.isManualImmediateAck) {
				if (!this.isManualAck) {
					updatePendingOffsets();
				}

So updatePendingOffsets() is only called when we have both !manualImmediateAck and !manualAck. Whereas in the processAck, we have this:

else {
				if (this.isManualImmediateAck) {
					try {
						ackImmediate(cRecord);
					}
					catch (@SuppressWarnings(UNUSED) WakeupException e) {
						// ignore - not polling
					}
				}
				else {
					addOffset(cRecord);
				}
			}

which means that we always call addOffset as long as !imanualImmediateAck including when ackManual is true. The way I understand this is that, we were always calling addOffset if the ack mode was not manual-immediate via the first call in handleAcks. But then, after this is done, in the processCommits method, we have the updatePendingOffset call, which checks for !manualAck in addition to !manualImmediateAck. But by now, the addOffset is already called via the handleAcks method. So, you are just cleaning up this second unnecessary call. Is that understanding correct? I want to ensure we are not introducing any regression by removing this logic.

[Wzy19930507]

So, you are just cleaning up this second unnecessary call. Is that understanding correct?

Yes! Exactly what I was want to say and more clear!