title: How to implement custom kafka Partition using spring cloud stream
body: <p>I am trying to implement a custom Kafka Partitioner using spring cloud stream bindings. I would like to just custom Partition the user topic and not do anything with company topic(Kafka will use DefaultPartitioner in this case).</p>
  
  <p>My bindings configuration:</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-css lang-css prettyprint-override"><code>spring:
    cloud:
      stream:
        bindings:
          comp-out:
            destination: company
            contentType: application/json
          user-out:
            destination: user
            contentType: application/json</code></pre>
  </div>
  </div>
  </p>
  
  <p>As per reference document: <a href="https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/2.1.0.RC4/single/spring-cloud-stream-binder-kafka.html#_partitioning_with_the_kafka_binder" rel="nofollow noreferrer">https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/2.1.0.RC4/single/spring-cloud-stream-binder-kafka.html#_partitioning_with_the_kafka_binder</a>
  I modified the configuration to this:</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-css lang-css prettyprint-override"><code>spring:
    cloud:
      stream:
        bindings:
          comp-out:
            destination: company
            contentType: application/json
          user-out:
            destination: user
            contentType: application/json
          producer:
              partitioned: true
              partitionSelectorClass: config.UserPartitioner</code></pre>
  </div>
  </div>
  </p>
  
  <p>I Post the message into Stream using this:</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-js lang-js prettyprint-override"><code>public void postUserStream(User user) throws ServiceException {
          try {
              LOG.info("Posting User {} into Kafka stream...", user);
              MessageChannel messageChannel = messageStreams.outboundUser();
              messageChannel
                      .send(MessageBuilder.withPayload(user)
                              .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build());
          } catch (Exception ex) {
              LOG.error("Error while populating User stream into Kafka.. ", ex);
              throw ex;
          }
      }</code></pre>
  </div>
  </div>
  </p>
  
  <p>My UserPartitioner Class:</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-js lang-js prettyprint-override"><code>public class UserPartitioner extends DefaultPartitioner {
      @Override
      public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes,
            Cluster cluster) {
  
          String partitionKey = null;
          if (Objects.nonNull(value)) {
              User user = (User) value;
              partitionKey = String.valueOf(user.getCompanyId()) + "_" + String.valueOf(user.getId());
              keyBytes = partitionKey.getBytes();
          }
          return super.partition(topic, partitionKey, keyBytes, value, valueBytes, cluster);
      }
  }</code></pre>
  </div>
  </div>
  </p>
  
  <p>I end up receiving following exception:</p>
  
  <p>Description:
  Failed to bind properties under 'spring.cloud.stream.bindings.user-out.producer' to org.springframework.cloud.stream.binder.ProducerProperties:</p>
  
  <pre><code>Property: spring.cloud.stream.bindings.user-out.producer.partitioned
  Value: true
  Origin: "spring.cloud.stream.bindings.user-out.producer.partitioned" from property source "bootstrapProperties"
  Reason: No setter found for property: partitioned
  </code></pre>
  
  <p>Action:
  Update your application's configuration</p>
  
  <p>Any reference link on how to set up Custom Partition using message binders will be helpful.</p>
  
  <p>Edit: Based on the documentation Tried the below steps as well:</p>
  
  <p>user-out:
            destination: user
            contentType: application/json
            producer:
             partitionKeyExtractorClass: config.SimpleUserPartitioner</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-js lang-js prettyprint-override"><code>@Component
  public class SimpleUserPartitioner implements PartitionKeyExtractorStrategy {
  
   @Override
   public Object extractKey(Message&lt;?&gt; message) {
    if(message.getPayload() instanceof BaseUser) {
     BaseUser user = (BaseUser) message.getPayload();
     return user.getId();
    }
    return 10;
   }
      
  }</code></pre>
  </div>
  </div>
  </p>
  
  <p>update 2: Solution that worked for me add partitioncount to bindings and autoaddpartitions to true in binder:</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-css lang-css prettyprint-override"><code>spring:
    logging:
      level: info
    cloud:
      stream:
        bindings:
          user-out:
            destination: user
            contentType: application/json
            producer:
              partition-key-expression: headers['partitionKey']
              partition-count: 4</code></pre>
  </div>
  </div>
  </p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-css lang-css prettyprint-override"><code>spring:
    cloud:
      stream:
        kafka:
          binder:
            brokers: localhost:9092
            autoAddPartitions: true</code></pre>
  </div>
  </div>
  </p>