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

INTEXT-206: Don't create extra consumers

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-206

The previous logic in the `ConcurrentMessageListenerDispatcher` create exactly `consumers`
number of `QueueingMessageListenerInvoker`, although we use only the `partitions.size()`.

Fix the logic to the `this.consumers = Math.min(partitions.size(), consumers);` to create maximum
consumers according to the partitions count.
This commit is contained in:
Artem Bilan
2015-10-26 21:13:14 -04:00
committed by Artem Bilan
parent 464ca70aa6
commit 28aca5332e
1 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java
View File

@@ -20,18 +20,14 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import com.gs.collections.api.block.procedure.Procedure;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.Partition;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.Assert;
import com.gs.collections.api.block.procedure.Procedure;
import com.gs.collections.api.block.procedure.Procedure2;
import com.gs.collections.api.map.MutableMap;
import com.gs.collections.impl.factory.Maps;
@@ -42,6 +38,7 @@ import com.gs.collections.impl.factory.Maps;
* from the same partition are being processed in their original order.
*
* @author Marius Bogoevici
* @author Artem Bilan
*/
class ConcurrentMessageListenerDispatcher {
@@ -72,7 +69,8 @@ class ConcurrentMessageListenerDispatcher {
private MutableMap<Partition, QueueingMessageListenerInvoker> delegates;
public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler,
Collection<Partition> partitions, OffsetManager offsetManager, int consumers, int queueSize, Executor taskExecutor) {
Collection<Partition> partitions, OffsetManager offsetManager,
int consumers, int queueSize, Executor taskExecutor) {
Assert.isTrue
(delegateListener instanceof MessageListener
|| delegateListener instanceof AcknowledgingMessageListener,
@@ -86,7 +84,7 @@ class ConcurrentMessageListenerDispatcher {
this.errorHandler = errorHandler;
this.partitions = partitions;
this.offsetManager = offsetManager;
this.consumers = consumers;
this.consumers = Math.min(partitions.size(), consumers);
this.queueSize = queueSize;
this.taskExecutor = taskExecutor;
}
@@ -121,7 +119,8 @@ class ConcurrentMessageListenerDispatcher {
List<QueueingMessageListenerInvoker> delegateList = new ArrayList<QueueingMessageListenerInvoker>(consumers);
for (int i = 0; i < consumers; i++) {
QueueingMessageListenerInvoker queueingMessageListenerInvoker =
new QueueingMessageListenerInvoker(queueSize, offsetManager, delegateListener, errorHandler, taskExecutor);
new QueueingMessageListenerInvoker(queueSize, offsetManager, delegateListener, errorHandler,
taskExecutor);
delegateList.add(queueingMessageListenerInvoker);
}