title: Kafka - not all consumers receive subscribed message
body: <p>To generically publish messages using Kafka, I'm using the class name as the topic:</p>
  
  <pre><code>kafkaProducer.send(new ProducerRecord(object.getClass().getName(), new DomainObjectAdapter(object).toJsonString()));
  </code></pre>
  
  <p>And the consumers subscribe to the classes they are interested in:</p>
  
  <pre><code>    for(Object sub:_subscriptions)
          topics.add(sub.getClass().getName());
      _kafkaConsumer.subscribe(topics);
  </code></pre>
  
  <p>The problem is, only one of the consumers ever receive the subscribed to message.  My understanding is that kafka will assign a unique partition to each subscriber (if available).  I currently only have 2 subscribers and my kafka server.properties specified 4 partitions.  It appears that all consumers are reading from the same partition.  Perhaps Kafka is a poor choice for a service bus due to this apparent limitation.  Any help would be much appreciated!</p>
  
  <p>Kafka consumer properties:</p>
  
  <pre><code>    properties.put("bootstrap.servers", _settings.getEndpoint());
      properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      properties.put("enable.auto.commit", "false");
      properties.put("group.id", "TestGroup");
      properties.put("auto.offset.reset","earliest");
  </code></pre>
  
  <p>Kafka producer properties:</p>
  
  <pre><code>    properties.put("bootstrap.servers",_settings.getEndpoint());
      properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  </code></pre>
  
  <p>Server properties (the only thing I changed from the default properties):</p>
  
  <pre><code>num.partitions=4
  </code></pre>
  
  <p>Note: I've also tried consumer settings as:</p>
  
  <pre><code>    properties.put("bootstrap.servers", _settings.getEndpoint());
      properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      properties.put("auto.commit.interval.ms","1000");
      properties.put("enable.auto.commit", "true");
      properties.put("group.id", "testGroup");
      properties.put("auto.offset.reset","latest");
      properties.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
  </code></pre>