Skip to content

Commit 654c531

Browse files
tsalsosobychacko
authored andcommitted
GH-2246: Partition attribute SpEL expression
Fixes: gh-2246 When partition attribute is specified as a SpEL expression on PartitionOffset annotation, it doesn not resolve the value from the expression. Fixing this issue. Adding test to verify. **Cherry-pick to 3.0.x**
1 parent 3314e79 commit 654c531

File tree

2 files changed

+17
-2
lines changed

2 files changed

+17
-2
lines changed

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -867,7 +867,7 @@ private List<TopicPartitionOffset> resolveTopicPartitionsList(TopicPartition top
867867
for (String partition : partitions) {
868868
resolvePartitionAsInteger((String) topic, resolveExpression(partition), result, null, false, false);
869869
}
870-
if (partitionOffsets.length == 1 && partitionOffsets[0].partition().equals("*")) {
870+
if (partitionOffsets.length == 1 && resolveExpression(partitionOffsets[0].partition()).equals("*")) {
871871
result.forEach(tpo -> {
872872
tpo.setOffset(resolveInitialOffset(tpo.getTopic(), partitionOffsets[0]));
873873
tpo.setRelativeToCurrent(isRelative(tpo.getTopic(), partitionOffsets[0]));

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -175,10 +175,11 @@
175175
* @author Venil Noronha
176176
* @author Dimitri Penner
177177
* @author Nakul Mishra
178+
* @author Soby Chacko
178179
*/
179180
@SpringJUnitConfig
180181
@DirtiesContext
181-
@EmbeddedKafka(topics = { "annotated1", "annotated2", "annotated3",
182+
@EmbeddedKafka(topics = { "annotated1", "annotated2", "annotated3", "annotated3x",
182183
"annotated4", "annotated5", "annotated6", "annotated7", "annotated8", "annotated8reply",
183184
"annotated9", "annotated10",
184185
"annotated11", "annotated12", "annotated13", "annotated14", "annotated15", "annotated16", "annotated17",
@@ -311,6 +312,10 @@ public void manyTests() throws Exception {
311312
assertThat(this.listener.capturedRecord.value()).isEqualTo("foo");
312313
assertThat(this.config.listen3Exception).isNotNull();
313314

315+
template.send("annotated3x", 0, "foo");
316+
assertThat(this.listener.latch3x.await(60, TimeUnit.SECONDS)).isTrue();
317+
assertThat(this.listener.capturedRecord.value()).isEqualTo("foo");
318+
314319
template.send("annotated4", 0, "foo");
315320
assertThat(this.listener.latch4.await(60, TimeUnit.SECONDS)).isTrue();
316321
assertThat(this.listener.capturedRecord.value()).isEqualTo("foo");
@@ -1832,6 +1837,8 @@ static class Listener implements ConsumerSeekAware {
18321837

18331838
final CountDownLatch latch3 = new CountDownLatch(1);
18341839

1840+
final CountDownLatch latch3x = new CountDownLatch(1);
1841+
18351842
final CountDownLatch latch4 = new CountDownLatch(1);
18361843

18371844
final CountDownLatch latch5 = new CountDownLatch(1);
@@ -2010,6 +2017,14 @@ public void listen3(ConsumerRecord<?, ?> record) {
20102017
this.latch3.countDown();
20112018
}
20122019

2020+
@KafkaListener(id = "partitionExpression", topicPartitions = @TopicPartition(topic = "${topicThree:annotated3x}",
2021+
partitions = "${zero:0}",
2022+
partitionOffsets = @PartitionOffset(partition = "#{'*'}", initialOffset = "0")))
2023+
public void listenPartitionSpelExpression(ConsumerRecord<?, ?> record) {
2024+
this.capturedRecord = record;
2025+
this.latch3x.countDown();
2026+
}
2027+
20132028
@KafkaListener(id = "#{'qux'}", topics = "annotated4",
20142029
containerFactory = "kafkaManualAckListenerContainerFactory", containerGroup = "qux#{'Group'}",
20152030
properties = {

0 commit comments

Comments
 (0)