Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3637: Fix KafkaMessageSource First Poll

Browse Source 
Resolves https://github.com/spring-projects/spring-integration/issues/3637

If there are no records to receive, the poll blocked for `20*pollTimeout`.

Wake the consumer during partition assignment; however, this changes the behavior
when there are records present - the first poll always returns no records.

Detect that the wakeup was due to assignment and perform another poll.

Also add some debug logging, and don't create a new consumer if the source was
stopped.

**cherry-pick to 5.4.x**
This commit is contained in:
Gary Russell
2021-09-30 15:19:36 -04:00
committed by Artem Bilan
parent 50c4304b03
commit 8a5c0377ef
1 changed files with 31 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
View File

@@ -139,6 +139,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
private Duration closeTimeout = Duration.ofSeconds(DEFAULT_CLOSE_TIMEOUT);
public boolean newAssignment;
private volatile Consumer<K, V> consumer;
private volatile boolean pausing;
@@ -147,6 +149,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
private volatile Iterator<ConsumerRecord<K, V>> recordsIterator;
private volatile boolean stopped;
/**
* Construct an instance with the supplied parameters. Fetching multiple
* records per poll will be disabled.
@@ -387,12 +391,14 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
@Override
public synchronized void start() {
this.running = true;
this.stopped = false;
}
@Override
public synchronized void stop() {
stopConsumer();
this.running = false;
this.stopped = true;
}
@Override
@@ -412,6 +418,10 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
@Override
protected synchronized Object doReceive() {
if (this.stopped) {
this.logger.debug("Message source is stopped; no records will be returned");
return null;
}
if (this.consumer == null) {
createConsumer();
this.running = true;
@@ -512,14 +522,27 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
}
else {
synchronized (this.consumerMonitor) {
ConsumerRecords<K, V> records = this.consumer
.poll(this.assignedPartitions.isEmpty() ? this.assignTimeout : this.pollTimeout);
if (records == null || records.count() == 0) {
try {
ConsumerRecords<K, V> records = this.consumer
.poll(this.assignedPartitions.isEmpty() ? this.assignTimeout : this.pollTimeout);
this.logger.debug(() -> records == null
? "Received null"
: "Received " + records.count() + " records");
if (records == null || records.count() == 0) {
return null;
}
this.remainingCount.set(records.count());
this.recordsIterator = records.iterator();
return nextRecord();
}
catch (WakeupException ex) {
this.logger.debug("Woken");
if (this.newAssignment) {
this.newAssignment = false;
return pollRecord();
}
return null;
}
this.remainingCount.set(records.count());
this.recordsIterator = records.iterator();
return nextRecord();
}
}
}
@@ -637,6 +660,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
this.providedRebalanceListener.onPartitionsAssigned(partitions);
}
}
KafkaMessageSource.this.consumer.wakeup();
KafkaMessageSource.this.newAssignment = true;
}
}