title: Custom partitioning with Java kafka client
body: <p>I was able to write a kafka sample app in Java. It has 3 topics and pub/sub is working fine. But couldn't assign those topics into different partitions. </p>
  
  <p><strong>My Consumer</strong></p>
  
  <pre><code>public class Consumers extends Thread {
      private static final List&lt;String&gt; TOPIC_LIST = Arrays.asList("topic1", "topic2", "topic3");
      private static final List&lt;TopicPartition&gt; PARTITION_LIST = 
  Arrays.asList(new TopicPartition(TOPIC_LIST.get(0), 1), new TopicPartition(TOPIC_LIST.get(1), 2));
  
      private void message() {
          Properties consumerProperties = KafkaProperties.getConsumerProperties();
          org.apache.kafka.clients.consumer.KafkaConsumer&lt;String, String&gt; consumer = new KafkaConsumer&lt;&gt;(consumerProperties);
          consumer.assign(PARTITION_LIST);
          Logger.debug("Kafka IP : " + consumerProperties.getProperty("bootstrap.servers"));
          try {
              while (true) {
                  ConsumerRecords&lt;String, String&gt; records = consumer.poll(100);
                  for (ConsumerRecord&lt;String, String&gt; record : records) {
                      process(record.topic(), record.value());
                  }
              }
          } catch (Exception e) {
              Logger.error("error while consuming : " + e.getMessage());
              e.printStackTrace();
          } finally {
              consumer.close();
          }
      }
  
      private void process(String topic, String value) {
          KafkaProcessor.process(topic, value);
      }
  
      @Override
      public void run() {
          message();
      }
  }
  </code></pre>
  
  <p><strong>My Custom Partitioner</strong></p>
  
  <pre><code>public class CustomPartitioner implements Partitioner {
  private static Map&lt;String, Integer&gt; partitionMap;
  
  @Override
  public void configure(Map&lt;String, ?&gt; configs) {
      System.out.println("Inside CustomPartitioner.configure " + configs);
      partitionMap = new HashMap&lt;&gt;();
      for (Map.Entry&lt;String, ?&gt; entry : configs.entrySet()) {
          if (entry.getKey().startsWith("partitions.")) {
              String keyName = entry.getKey();
              String value = (String) entry.getValue();
              int partitionId = Integer.parseInt(keyName.substring(11));
              partitionMap.put(value, partitionId);
          }
      }
  }
  
  @Override
  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
      List partitions = cluster.availablePartitionsForTopic(topic);
      String valueStr = (String) value;
      String name = ((String) value).split(":")[0];
      if (partitionMap.containsKey(name)) {
          //If the country is mapped to particular partition return it
          return partitionMap.get(name);
      } else {
          //If no country is mapped to particular partition distribute between remaining partitions
          int noOfPartitions = cluster.topics().size();
          return value.hashCode() % noOfPartitions + partitionMap.size();
      }
  }
  
  public void close() {
  }}
  </code></pre>
  
  <p><strong>My Producer</strong></p>
  
  <pre><code>public void producer(String topic, String message) {
      Producer&lt;String, String&gt; producer = new KafkaProducer&lt;&gt;(KafkaProperties.getProducerProperties());
      try {
          ProducerRecord&lt;String, String&gt; producerRecord = new ProducerRecord&lt;&gt;(topic, null, message);
          producer.send(producerRecord);
          producer.close();
      } catch (Exception e) {
          Logger.error("kafka message publish error: ", e);
      }
  }
  </code></pre>
  
  <p>When I run this code. I've got following warnings.</p>
  
  <pre><code>[warn] o.a.k.c.p.ProducerConfig - The configuration partitions.2 = partition2 was supplied but isn't a known config.
  [warn] o.a.k.c.p.ProducerConfig - The configuration partitions.1 = partition1 was supplied but isn't a known config.
  </code></pre>
  
  <p>My producer properties are as follows,</p>
  
  <pre><code> properties.put("bootstrap.servers", "127.0.0.1:9092");
   properties.put("acks", "all");
   properties.put("retries", 0);
   properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class.getCanonicalName());
   properties.put("partitions.1", "partition1");
   properties.put("partitions.2", "partition2");
   properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
   properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  </code></pre>
  
  <p>With these properties and code I couldn't send or receive messages. How may I fix this.</p>