title: Spring XD very poor performance when using redis as transport
body: <p>I deployed an distributed configuration on my local machine. </p>
  
  <p>1 admin
  1 container
  1 zookeeper
  1 hsql
  1 redis</p>
  
  <p>I have the follwing streams:</p>
  
  <pre><code>stream create --name activityLog --definition "tail --name=/home/bruno/randomName | log"
  stream create --name activityLogCounterTap --definition "tap:stream:activityLog &gt; json-to-tuple | filter --expression=payload.reportable.equals(true) | counter --name=activitylogcount"
  stream create --name activityLogEventTypeCounterTap --definition "tap:stream:activityLogCounterTap.filter &gt; field-value-counter --fieldName=status --name=status"
  stream create --name activityLogClientTypeCounterTap --definition "tap:stream:activityLogCounterTap.filter &gt; field-value-counter --fieldName=clientType --name=clientType"
  stream create --name activityLogIndexCounterTap --definition "tap:stream:activityLogCounterTap.filter &gt; field-value-counter --fieldName=index --name=index"
  stream create --name activityLogCustomerCounterTap --definition "tap:stream:activityLogCounterTap.filter &gt; field-value-counter --fieldName=customer --name=customer"
  stream create --name activityLogChannelCounterTap --definition "tap:stream:activityLogCounterTap.filter &gt; field-value-counter --fieldName=channel --name=channel"
  stream create --name activityLogStrategyOnlineCounterTap --definition "tap:stream:activityLogCounterTap.filter &gt; field-value-counter --fieldName=strategyOnline --name=strategyOnline"
  stream create --name activityLogStrategyOfflineCounterTap --definition "tap:stream:activityLogCounterTap.filter &gt; field-value-counter --fieldName=strategyOffline --name=strategyOffline"
  </code></pre>
  
  <p>I want to deploy using the following deployment descriptor:</p>
  
  <pre><code>stream deploy --name activityLog --properties "module.*.count=0"
  stream deploy --name activityLogCounterTap --properties "module.*.count=0"
  stream deploy --name activityLogEventTypeCounterTap --properties "module.*.count=0"
  stream deploy --name activityLogClientTypeCounterTap --properties "module.*.count=0"
  stream deploy --name activityLogIndexCounterTap --properties "module.*.count=0"
  stream deploy --name activityLogCustomerCounterTap --properties "module.*.count=0"
  stream deploy --name activityLogChannelCounterTap --properties "module.*.count=0"
  stream deploy --name activityLogStrategyOnlineCounterTap --properties "module.*.count=0"
  stream deploy --name activityLogStrategyOfflineCounterTap --properties "module.*.count=0"
  </code></pre>
  
  <p>The messages are json objects.</p>
  
  <p>The idea is to have an arbitraty number for containers consuming messages from apache kafka however for the example a simple tail is used.</p>
  
  <p>I deployed one by one and I can see that the performance using redis as transport degrades a lot after the 4th stream is deployed. I observe using visual vm that there are some sort of race conditions in inbound.*redis:queue-inbound-channel-adapter1. If I deploy more streams(taps) it becomes unusable.</p>
  
  <p>(I tried to add an image of threads view from Visual VM here however I need 10 reputation :s)</p>
  
  <p>It seems to be locking here:</p>
  
  <pre><code>"inbound.activityLog.0-redis:queue-inbound-channel-adapter1" prio=10 tid=0x00007fe3581a1800 nid=0x29c7 waiting on condition [0x00007fe3d28de000]
     java.lang.Thread.State: WAITING (parking)
      at sun.misc.Unsafe.park(Native Method)
      - parking to wait for  &lt;0x0000000788b48d48&gt; (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
      at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
      at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
      at org.apache.commons.pool2.impl.LinkedBlockingDeque.takeFirst(LinkedBlockingDeque.java:524)
      at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:438)
      at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:361)
      at redis.clients.util.Pool.getResource(Pool.java:40)
      at org.springframework.data.redis.connection.jedis.JedisConnectionFactory.fetchJedisConnector(JedisConnectionFactory.java:90)
      at org.springframework.data.redis.connection.jedis.JedisConnectionFactory.getConnection(JedisConnectionFactory.java:143)
      at org.springframework.data.redis.connection.jedis.JedisConnectionFactory.getConnection(JedisConnectionFactory.java:41)
      at org.springframework.data.redis.core.RedisConnectionUtils.doGetConnection(RedisConnectionUtils.java:128)
      at org.springframework.data.redis.core.RedisConnectionUtils.getConnection(RedisConnectionUtils.java:91)
      at org.springframework.data.redis.core.RedisConnectionUtils.getConnection(RedisConnectionUtils.java:78)
      at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:177)
      at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:152)
      at org.springframework.data.redis.core.AbstractOperations.execute(AbstractOperations.java:84)
      at org.springframework.data.redis.core.DefaultListOperations.rightPop(DefaultListOperations.java:151)
      at org.springframework.data.redis.core.DefaultBoundListOperations.rightPop(DefaultBoundListOperations.java:92)
      at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.popMessageAndSend(RedisQueueMessageDrivenEndpoint.java:177)
      at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.access$300(RedisQueueMessageDrivenEndpoint.java:50)
      at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint$ListenerTask.run(RedisQueueMessageDrivenEndpoint.java:290)
      at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52)
      at java.lang.Thread.run(Thread.java:745)
  
     Locked ownable synchronizers:
      - None
  </code></pre>
  
  <p>If less than 4 streams are deployed the waiting condition never happens:</p>
  
  <pre><code>"inbound.activityLog.0-redis:queue-inbound-channel-adapter1" prio=10 tid=0x00007ff5101f6800 nid=0x352c runnable [0x00007ff5794a4000]
     java.lang.Thread.State: RUNNABLE
      at java.net.SocketInputStream.socketRead0(Native Method)
      at java.net.SocketInputStream.read(SocketInputStream.java:152)
      at java.net.SocketInputStream.read(SocketInputStream.java:122)
      at java.net.SocketInputStream.read(SocketInputStream.java:108)
      at redis.clients.util.RedisInputStream.fill(RedisInputStream.java:109)
      at redis.clients.util.RedisInputStream.readByte(RedisInputStream.java:45)
      at redis.clients.jedis.Protocol.process(Protocol.java:120)
      at redis.clients.jedis.Protocol.read(Protocol.java:191)
      at redis.clients.jedis.Connection.getBinaryMultiBulkReply(Connection.java:212)
      at redis.clients.jedis.BinaryJedis.brpop(BinaryJedis.java:2068)
      at org.springframework.data.redis.connection.jedis.JedisConnection.bRPop(JedisConnection.java:1514)
      at org.springframework.data.redis.core.DefaultListOperations$12.inRedis(DefaultListOperations.java:154)
      at org.springframework.data.redis.core.AbstractOperations$ValueDeserializingRedisCallback.doInRedis(AbstractOperations.java:50)
      at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:190)
      at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:152)
      at org.springframework.data.redis.core.AbstractOperations.execute(AbstractOperations.java:84)
      at org.springframework.data.redis.core.DefaultListOperations.rightPop(DefaultListOperations.java:151)
      at org.springframework.data.redis.core.DefaultBoundListOperations.rightPop(DefaultBoundListOperations.java:92)
      at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.popMessageAndSend(RedisQueueMessageDrivenEndpoint.java:177)
      at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint.access$300(RedisQueueMessageDrivenEndpoint.java:50)
      at org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint$ListenerTask.run(RedisQueueMessageDrivenEndpoint.java:290)
      at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52)
      at java.lang.Thread.run(Thread.java:745)
  
     Locked ownable synchronizers:
      - None
  </code></pre>
  
  <ul>
  <li><p>My first question is why isn't spring-xd using redis as transport being able to handle more taps. I tried with rabbitMq and it works.</p></li>
  <li><p>My second question is why is it using redis in the first place hence I deployed with in memory configuration. Again using in memory configuration but this time using rabbitMQ I can trace messages being passed which shouldn't happen I think.</p></li>
  </ul>
  
  <p>--edited--</p>
  
  <p>So regarding the first question I figured out maxTotal parameter in org.apache.commons.pool2.GenericObjectPool default is 8. I went in debug mode and altered this value in runtime (no limit = -1) and that seems to solve de problem. To be able to configure it to deploy with that hack I needed to create an extension (from documentation) and override the following beans:</p>
  
  <pre><code>&lt;bean id="messageBus" class="org.springframework.xd.dirt.integration.redis.RedisMessageBus"&gt;
              &lt;constructor-arg ref="redisConnectionFactory2" /&gt;
              &lt;constructor-arg ref="codec"/&gt;
              &lt;property name="defaultBackOffInitialInterval" value="${xd.messagebus.redis.default.backOffInitialInterval}" /&gt;
              &lt;property name="defaultBackOffMaxInterval" value="${xd.messagebus.redis.default.backOffMaxInterval}" /&gt;
              &lt;property name="defaultBackOffMultiplier" value="${xd.messagebus.redis.default.backOffMultiplier}" /&gt;
              &lt;property name="defaultConcurrency" value="${xd.messagebus.redis.default.concurrency}" /&gt;
              &lt;property name="defaultMaxAttempts" value="${xd.messagebus.redis.default.maxAttempts}" /&gt;
      &lt;/bean&gt;
  
  
  &lt;bean id="counterRepository"
              class="org.springframework.xd.analytics.metrics.redis.RedisCounterRepository"&gt;
              &lt;constructor-arg ref="redisConnectionFactory2" /&gt;
      &lt;/bean&gt;
  
      &lt;bean id="fieldValueCounterRepository"
              class="org.springframework.xd.analytics.metrics.redis.RedisFieldValueCounterRepository"&gt;
              &lt;constructor-arg ref="redisConnectionFactory2" /&gt;
      &lt;/bean&gt;
  
      &lt;bean id="gaugeRepository"
              class="org.springframework.xd.analytics.metrics.redis.RedisGaugeRepository"&gt;
              &lt;constructor-arg ref="redisConnectionFactory2" /&gt;
      &lt;/bean&gt;
  
      &lt;bean id="richGaugeRepository"
              class="org.springframework.xd.analytics.metrics.redis.RedisRichGaugeRepository"&gt;
              &lt;constructor-arg ref="redisConnectionFactory2" /&gt;
      &lt;/bean&gt;
  
      &lt;bean id="aggregateCounterRepository"
              class="org.springframework.xd.analytics.metrics.redis.RedisAggregateCounterRepository"&gt;
              &lt;constructor-arg ref="redisConnectionFactory2" /&gt;
      &lt;/bean&gt;
  
  &lt;bean id="redisConnectionFactory2" lazy-init="false"
        class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"&gt;
      &lt;property name="hostName" value="${spring.redis.host}" /&gt;
      &lt;property name="port" value="${spring.redis.port}" /&gt;
      &lt;property name="password" value="" /&gt;
      &lt;property name="poolConfig" ref="jedisPoolConfig"/&gt;
  &lt;/bean&gt;
  
  &lt;bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig"&gt;
      &lt;property name="maxTotal" value="-1"/&gt;
  &lt;/bean&gt;
  </code></pre>
  
  <p>I tried at first just to define a redisConnectionFactory bean that would override the default one however it didn't work that way.</p>
  
  <p>I'm not sure if the solution is to make the maxTotal configurable within servers.yml to allow the increase of that value or if somehow there is a bug in which the poolables are not being released. Anyway I will open a jira an hope this helps someone not going through all the work I went :-)</p>