GH-2220: Fix TopicPartitionOffset for Retry Topics #2223

tomazfernandes · 2022-04-08T00:33:26Z

Resolves #2220

Fix TopicPartitionOffset for Retry Topics
Register NewTopics for Retry Topics only once per topic.
Add docs explaining configuration behavior for retry topics when more than one listener container is used for the same topic.

Please let me know if there's anything to be changed.

Thanks

artembilan

Here is my review, but I'm not fully sure what is going on from the logic perspective, so I'll leave that review part to @garyrussell .

Thanks

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java

tomazfernandes · 2022-04-08T16:07:18Z

Thanks a lot for the review @artembilan! I think I've addressed all the points, please let me know if I missed out on something, or if there's anything else.

artembilan

OK with me.

Let's give @garyrussell to review this as well on Monday!

garyrussell · 2022-04-13T15:43:24Z

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java

@@ -71,7 +74,12 @@ protected EndpointCustomizer addSuffixesAndMethod(DestinationTopic.Properties pr
 			Collection<EndpointCustomizer.TopicNamesHolder> topics = customizeAndRegisterTopics(namesProvider, endpoint);
 			endpoint.setId(namesProvider.getEndpointId(endpoint));
 			endpoint.setGroupId(namesProvider.getGroupId(endpoint));
-			endpoint.setTopics(topics.stream().map(EndpointCustomizer.TopicNamesHolder::getCustomizedTopic).toArray(String[]::new));
+			if (endpoint.getTopics().isEmpty() && endpoint.getTopicPartitionsToAssign() != null) {
+				endpoint.setTopicPartitions(getTopicPartitions(properties, namesProvider, endpoint.getTopicPartitionsToAssign()));


This breaks the initial offset on the main listener because it overwrites the existing ones with a null initial offset.

We shouldn't do this for the mainEndpoint in processAndRegisterEndpoint.

@SpringBootApplication public class Kgh2220Application { public static void main(String[] args) { SpringApplication.run(Kgh2220Application.class, args); } @Bean public NewTopic topic() { return TopicBuilder.name("kgh2220").partitions(1).replicas(1).build(); } @KafkaListener(id = "kgh2220xxx", topicPartitions = @TopicPartition(topic = "kgh2220", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0"))) @RetryableTopic void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, @Header(KafkaHeaders.OFFSET) long offset) { System.out.println(in + "@" + offset); System.out.println(topic); System.out.println(KafkaUtils.getConsumerGroupId()); // throw new RuntimeException(); } @Bean ApplicationRunner runner(KafkaTemplate<String, String> template) { return args -> { template.send("kgh2220", "foo"); }; } }

foo@15 kgh2220 kgh2220xxx

Good catch, makes sense. I'll fix it shortly, thanks.

Resolves spring-projects#2220 * Register NewTopics for Retry Topics only once per topic. * Add docs explaining configuration behavior for retry topics when more than one listener container is used for the same topic.

Make EndpointCustomizerFactory ctor public

garyrussell · 2022-04-13T18:55:27Z

For test inspiration - see this test (using mocks) that verifies the container performed the initial seeks when the listener was so configured.

spring-kafka/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentOnErrorRecordModeTests.java

Lines 99 to 103 in feb3563

    
           inOrder.verify(this.consumer).assign(any(Collection.class)); 
        
           inOrder.verify(this.consumer).seek(new TopicPartition("foo", 0), 0L); 
        
           inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 0L); 
        
           inOrder.verify(this.consumer).seek(new TopicPartition("foo", 2), 0L); 
        
           inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));

garyrussell · 2022-04-13T18:57:20Z

I guess you could just clone that test and add a @RetryableTopic to its listener.

tomazfernandes · 2022-04-13T19:00:43Z

Looks great, thanks a lot @garyrussell! Will finish up at least basic unit test coverage and try this out.

tomazfernandes · 2022-04-13T23:18:55Z

@garyrussell, I stumbled into something that might be a small bug in the TopicPartitionOffset code.

The hashCode method is:

@Override
public int hashCode() {
    return this.topicPartition.hashCode() + this.position.hashCode();
}

But position is a nullable field, so in this case it throws a NPE if used. If that's an actual bug, DYT I should fix this as part of this PR, or should I leave it alone? The obvious fix I see would be something like:

@Override
public int hashCode() {
    return this.position != null
                      ? this.topicPartition.hashCode() + this.position.hashCode()
                      : this.topicPartition.hashCode();
}

But if you have a different idea LMK.

Thanks.

tomazfernandes · 2022-04-13T23:23:16Z

Or maybe add the other fields to the calculation if they're also not null.

tomazfernandes · 2022-04-14T00:13:22Z

Also, the equals method doesn't take the offset into account:

	@Override
	public boolean equals(Object o) {
		if (this == o) {
			return true;
		}
		if (o == null || getClass() != o.getClass()) {
			return false;
		}
		TopicPartitionOffset that = (TopicPartitionOffset) o;
		return Objects.equals(this.topicPartition, that.topicPartition)
				&& Objects.equals(this.position, that.position);
	}

I'm catching these due to unexpected results in assertThat().equals().

tomazfernandes · 2022-04-14T01:11:53Z

@garyrussell, I actually can't really simply not call the endpoint customizer for the main topic because currently it's tied to the registration of the topics in the DestinationTopicProcessor, so I made it return early.

Although, what happens there today is that we use a no-ops suffixer for the main endpoint, so we basically get and set the same thing. But since users can provide their own implementation, I'm not sure that's not a breaking change - WDYT?

If that's a problem I can rework that to only leave the topic partition part out.

Thanks.

artembilan · 2022-04-14T14:27:21Z

How about this:

@Override
	public boolean equals(Object o) {
		if (this == o) return true;
		if (o == null || getClass() != o.getClass()) return false;
		TopicPartitionOffset that = (TopicPartitionOffset) o;
		return this.relativeToCurrent == that.relativeToCurrent &&
				Objects.equals(topicPartition, that.topicPartition)
				&& this.position == that.position
				&& Objects.equals(this.offset, that.offset);
	}

	@Override
	public int hashCode() {
		return Objects.hash(this.topicPartition, this.position, this.offset, this.relativeToCurrent);
	}

Just include everything into those methods.

Separate issue/ PR though...

I guess your tests must not rely on the equals() and hashCode() for now or at all and the verification logic must be slightly different.

garyrussell · 2022-04-14T15:36:09Z

I actually can't really simply not call the endpoint customizer for the main topic

Why not instead of

if (properties.isMainEndpoint()) {
	return topics;
}

use

if (endpoint.getTopics().isEmpty() && endpoint.getTopicPartitionsToAssign() != null) {
	endpoint.setTopicPartitions(getTopicPartitions(properties, namesProvider, endpoint.getTopicPartitionsToAssign()));
}
else if (!properties.isMainEndpoint()) {
	endpoint.setTopics(topics.stream().map(EndpointCustomizer.TopicNamesHolder::getCustomizedTopic).toArray(String[]::new));
}

Or, do you mean the other updates as also no-ops?

tomazfernandes · 2022-04-14T15:44:06Z

I think the logic you propose would incur in the same error, wouldn't it? There's nothing to prevent the main endpoint from getting into the first if clause. Or am I missing something?

I think if we just make sure the main endpoint doesn't enter either the topics or the TPO parts we should be good and least breaking as possible. WDYT?

And yes, all operations there are (or should be by default) no-ops for the main endpoint, except the getTopics part in the beginning which creates the TopicNamesHolder instances that we need.

tomazfernandes · 2022-04-14T15:49:56Z

Sorry - customizeAndRegisterTopics in the beggining.

tomazfernandes · 2022-04-14T15:58:41Z

I'm thinking this:

	if (!properties.isMainEndpoint()) {
		if (endpoint.getTopics().isEmpty() && endpoint.getTopicPartitionsToAssign() != null) {
			endpoint.setTopicPartitions(getTopicPartitions(properties, namesProvider, endpoint.getTopicPartitionsToAssign()));
		}
		else {
			endpoint.setTopics(endpoint.getTopics().stream().map(namesProvider::getTopicName).toArray(String[]::new));
		}
	}

Notice I changed the setTopics part a bit too so that it won't get names from TPOs if there are any. Although it should not get there with such state.

tomazfernandes · 2022-04-14T16:08:50Z

Or better:

	if (endpoint.getTopics().isEmpty() && endpoint.getTopicPartitionsToAssign() != null && !properties.isMainEndpoint()) {
		endpoint.setTopicPartitions(getTopicPartitions(properties, namesProvider, endpoint.getTopicPartitionsToAssign()));
	}
	else {
		endpoint.setTopics(endpoint.getTopics().stream().map(namesProvider::getTopicName).toArray(String[]::new));
	}

This has the least changes and also lets users go through the topics logic with a custom namesProvider if they wish to do something even for the main endpoint topics.

Since we're on a schedule for the release, I'll submit this logic along with the test changes, but I can change any of it if you prefer.

Adjust tests Improve TPO and topics handling

garyrussell

OK; I ran this version with my Boot app and it now works as expected.

If @artembilan 's approval from last week still stands, we can merge this.

tomazfernandes · 2022-04-14T18:31:57Z

I've just made a small improvement where users can also use the names provider with the main topic for topic partitions. To my understanding that's a really odd and unlikely combination, so I'm ok with leaving that out if you prefer - or I can submit if you don't mind looking at this again.

garyrussell · 2022-04-14T18:32:51Z

I am confused about the hashcode change, though (reverted); it looks like a valid NPE.

garyrussell · 2022-04-14T18:33:31Z

Sure; feel free if you have another commit.

artembilan · 2022-04-14T18:36:24Z

I would suggest to make TopicPartitionOffset equals() and hashCode() in the separate PR which could be back-ported to all supported versions.

garyrussell · 2022-04-14T18:39:32Z

@artembilan

	@Override
	public int hashCode() {
		return Objects.hash(this.topicPartition, this.position, this.offset, this.relativeToCurrent);
	}

Offset and relativeToCurrent are mutable so are not included in the equals or hash.

#1554

tomazfernandes · 2022-04-14T18:40:16Z

Well, TBH, since I'm not that familiar with TopicPartitionOffset, I'm not so sure about the new logic - that's why left it out in the first place - didn't want to add complexity so close to the release. I thought of doing this:

	private static TopicPartitionOffset getTPOForMainTopic(RetryTopicNamesProviderFactory.RetryTopicNamesProvider namesProvider, TopicPartitionOffset tpo) {
		return new TopicPartitionOffset(namesProvider.getTopicName(tpo.getTopic()),
				tpo.getPartition(), tpo.getOffset(), tpo.getPosition());
	}

But I'm not so sure that's an exact copy of the original just with the topic name potentially changed (if the user provides a names provider) - there seems to be a relativeToCurrent flag which is not considered in this logic.

Any pointers? As I said, I'm ok with leaving the logic as is too.

garyrussell · 2022-04-14T18:40:32Z

Good point (re fixing hash for back porting)

tomazfernandes · 2022-04-14T18:47:33Z

Well, I guess this should do the trick:

	private static TopicPartitionOffset getTPOForMainTopic(RetryTopicNamesProviderFactory.RetryTopicNamesProvider namesProvider, TopicPartitionOffset tpo) {
		TopicPartitionOffset newTpo = new TopicPartitionOffset(namesProvider.getTopicName(tpo.getTopic()),
				tpo.getPartition(), tpo.getOffset(), tpo.getPosition());
		newTpo.setRelativeToCurrent(tpo.isRelativeToCurrent());
		return newTpo;
	}

garyrussell · 2022-04-14T18:50:43Z

I am not sure why you want to replace the existing TPOs with an exact copy.

garyrussell · 2022-04-14T18:51:13Z

Oh - it's so they can change the name; OK; that makes sense.

@SuppressWarnings

* GH-2220: Fix TopicPartitionOffset for Retry Topics Resolves #2220 * Register NewTopics for Retry Topics only once per topic. * Add docs explaining configuration behavior for retry topics when more than one listener container is used for the same topic. * Move default partition to constant field. * Address review comments Make EndpointCustomizerFactory ctor public * Remove main endpoint customization * Remove failing @SuppressWarnings * Revert TPO hashCode changes Adjust tests Improve TPO and topics handling * Enable customizing main endpoint TPOs

@SuppressWarnings

* GH-2220: Fix TopicPartitionOffset for Retry Topics Resolves #2220 * Register NewTopics for Retry Topics only once per topic. * Add docs explaining configuration behavior for retry topics when more than one listener container is used for the same topic. * Move default partition to constant field. * Address review comments Make EndpointCustomizerFactory ctor public * Remove main endpoint customization * Remove failing @SuppressWarnings * Revert TPO hashCode changes Adjust tests Improve TPO and topics handling * Enable customizing main endpoint TPOs

@SuppressWarnings

* GH-2220: Fix TopicPartitionOffset for Retry Topics Resolves #2220 * Register NewTopics for Retry Topics only once per topic. * Add docs explaining configuration behavior for retry topics when more than one listener container is used for the same topic. * Move default partition to constant field. * Address review comments Make EndpointCustomizerFactory ctor public * Remove main endpoint customization * Remove failing @SuppressWarnings * Revert TPO hashCode changes Adjust tests Improve TPO and topics handling * Enable customizing main endpoint TPOs

garyrussell · 2022-04-14T19:20:19Z

...and cherry picked to 2.9.x, 2.8.x, and to 2.7.x after resolving minor conflicts.

garyrussell · 2022-04-14T19:22:32Z

Hmmm - test failures on the back ports...

org.mockito.exceptions.verification.WantedButNotInvoked: Wanted but not invoked:
consumerRecordMessage.topic();
-> at org.springframework.kafka.retrytopic.RetryTopicConfigurerTests.shouldLogConsumerRecordMessage(RetryTopicConfigurerTests.java:369)
Actually, there were zero interactions with this mock.

Wanted but not invoked:
consumerRecordMessage.topic();
-> at org.springframework.kafka.retrytopic.RetryTopicConfigurerTests.shouldLogConsumerRecordMessage(RetryTopicConfigurerTests.java:369)
(88 more lines...)

Investigating...

tomazfernandes · 2022-04-14T19:26:06Z

I changed these tests because they failed depending on the log level set for the tests. This is probably due to the change to KafkaUtils for logging. Easiest fix should be finding out which method is called on the consumer record and changing to assert that.

garyrussell · 2022-04-14T19:35:14Z

Thanks; fixed.

artembilan requested changes Apr 8, 2022

View reviewed changes

artembilan approved these changes Apr 8, 2022

View reviewed changes

garyrussell suggested changes Apr 13, 2022

View reviewed changes

tomazfernandes added 3 commits April 13, 2022 14:19

spring-projectsGH-2220: Fix TopicPartitionOffset for Retry Topics

a99832a

Resolves spring-projects#2220 * Register NewTopics for Retry Topics only once per topic. * Add docs explaining configuration behavior for retry topics when more than one listener container is used for the same topic.

Move default partition to constant field.

26914b3

Address review comments

6b5077a

Make EndpointCustomizerFactory ctor public

garyrussell mentioned this pull request Apr 13, 2022

GH-2226: Add RetryTopicConfigurationSupport #2227

Closed

tomazfernandes force-pushed the GH-2220 branch from a85975d to cf2105f Compare April 14, 2022 00:44

Remove main endpoint customization

aac62e6

tomazfernandes force-pushed the GH-2220 branch from cf2105f to aac62e6 Compare April 14, 2022 00:53

Remove failing @SuppressWarnings

eb1a0d6

Revert TPO hashCode changes

c78c6a2

Adjust tests Improve TPO and topics handling

garyrussell approved these changes Apr 14, 2022

View reviewed changes

artembilan approved these changes Apr 14, 2022

View reviewed changes

Enable customizing main endpoint TPOs

676314a

garyrussell approved these changes Apr 14, 2022

View reviewed changes

garyrussell merged commit 80b1770 into spring-projects:main Apr 14, 2022

GH-2220: Fix TopicPartitionOffset for Retry Topics #2223

GH-2220: Fix TopicPartitionOffset for Retry Topics #2223

Uh oh!

Conversation

tomazfernandes commented Apr 8, 2022

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tomazfernandes commented Apr 8, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

garyrussell Apr 13, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tomazfernandes Apr 13, 2022

Choose a reason for hiding this comment

Uh oh!

garyrussell commented Apr 13, 2022

Uh oh!

garyrussell commented Apr 13, 2022

Uh oh!

tomazfernandes commented Apr 13, 2022

Uh oh!

tomazfernandes commented Apr 13, 2022

Uh oh!

tomazfernandes commented Apr 13, 2022

Uh oh!

tomazfernandes commented Apr 14, 2022

Uh oh!

tomazfernandes commented Apr 14, 2022

Uh oh!

artembilan commented Apr 14, 2022

Uh oh!

garyrussell commented Apr 14, 2022

Uh oh!

tomazfernandes commented Apr 14, 2022

Uh oh!

tomazfernandes commented Apr 14, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

tomazfernandes commented Apr 14, 2022

Uh oh!

tomazfernandes commented Apr 14, 2022

Uh oh!

garyrussell left a comment

Choose a reason for hiding this comment

Uh oh!

tomazfernandes commented Apr 14, 2022

Uh oh!

garyrussell commented Apr 14, 2022

Uh oh!

garyrussell commented Apr 14, 2022

Uh oh!

artembilan commented Apr 14, 2022

Uh oh!

garyrussell commented Apr 14, 2022

Uh oh!

tomazfernandes commented Apr 14, 2022

Uh oh!

garyrussell commented Apr 14, 2022

Uh oh!

tomazfernandes commented Apr 14, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

garyrussell commented Apr 14, 2022

Uh oh!

garyrussell commented Apr 14, 2022

Uh oh!

garyrussell commented Apr 14, 2022

tomazfernandes commented Apr 8, 2022 •

edited

Loading

garyrussell Apr 13, 2022 •

edited

Loading

tomazfernandes commented Apr 14, 2022 •

edited

Loading

tomazfernandes commented Apr 14, 2022 •

edited

Loading