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

INTEXT-169: Ensure queueSize is a Power of 2

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

Error from lmax is too far removed from configuration.
This commit is contained in:
Gary Russell
2015-05-29 11:26:00 -04:00
committed by Artem Bilan
parent d475a79d83
commit 1b28a7b863
2 changed files with 36 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -31,23 +31,6 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import com.gs.collections.api.RichIterable;
import com.gs.collections.api.block.function.Function;
import com.gs.collections.api.block.predicate.Predicate;
import com.gs.collections.api.block.procedure.Procedure;
import com.gs.collections.api.block.procedure.Procedure2;
import com.gs.collections.api.collection.MutableCollection;
import com.gs.collections.api.list.ImmutableList;
import com.gs.collections.api.list.MutableList;
import com.gs.collections.api.multimap.MutableMultimap;
import com.gs.collections.api.partition.PartitionIterable;
import com.gs.collections.impl.block.factory.Functions;
import com.gs.collections.impl.block.function.checked.CheckedFunction;
import com.gs.collections.impl.factory.Lists;
import com.gs.collections.impl.factory.Multimaps;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.utility.Iterate;
import kafka.common.ErrorMapping;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -65,6 +48,25 @@ import org.springframework.integration.kafka.core.Result;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import com.gs.collections.api.RichIterable;
import com.gs.collections.api.block.function.Function;
import com.gs.collections.api.block.predicate.Predicate;
import com.gs.collections.api.block.procedure.Procedure;
import com.gs.collections.api.block.procedure.Procedure2;
import com.gs.collections.api.collection.MutableCollection;
import com.gs.collections.api.list.ImmutableList;
import com.gs.collections.api.list.MutableList;
import com.gs.collections.api.multimap.MutableMultimap;
import com.gs.collections.api.partition.PartitionIterable;
import com.gs.collections.impl.block.factory.Functions;
import com.gs.collections.impl.block.function.checked.CheckedFunction;
import com.gs.collections.impl.factory.Lists;
import com.gs.collections.impl.factory.Multimaps;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.utility.Iterate;
import kafka.common.ErrorMapping;
/**
* @author Marius Bogoevici
*/
@@ -243,9 +245,11 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
/**
* The maximum number of messages that are buffered by each concurrent {@link MessageListener} runner.
* Increasing the value may increase throughput, but also increases the memory consumption.
* Must be a power of 2.
* @param queueSize the queue size
*/
public void setQueueSize(int queueSize) {
Assert.isTrue(Integer.bitCount(queueSize) == 1, "'queueSize' must be a power of 2");
this.queueSize = queueSize;
}

18
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java
View File

@@ -18,9 +18,11 @@
package org.springframework.integration.kafka.listener;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
import static org.hamcrest.core.IsEqual.equalTo;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import java.util.ArrayList;
@@ -29,9 +31,6 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.gs.collections.api.multimap.list.MutableListMultimap;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
import org.junit.Rule;
import org.junit.Test;
@@ -47,6 +46,10 @@ import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import com.gs.collections.api.multimap.list.MutableListMultimap;
import com.gs.collections.impl.list.mutable.FastList;
import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap;
/**
* @author Marius Bogoevici
*/
@@ -79,6 +82,15 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
kafkaMessageListenerContainer.setMaxFetch(100);
kafkaMessageListenerContainer.setConcurrency(2);
try {
kafkaMessageListenerContainer.setQueueSize(1000);
fail("expected exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage(), containsString("power of 2"));
}
kafkaMessageListenerContainer.setQueueSize(1024);
int expectedMessageCount = 100;
final MutableListMultimap<Integer, KeyedMessageWithOffset> receivedData =