title: Kafka no partition allocation
body: <p>First Kafka project, so trying out simple programs. I wrote trivial Java programs to publish records to a Kafka topic and read records from the same topic. This is simple stuff, just to prove to myself that I know how to do it. The idea was to have a ReconcilePublish class, a ReconcileSubscribe class, and a Reconcile class that would use the other two to check that all the published messages were read, in the same order.</p>
  <p>Under certain circumstances, the ReconcileSubscribe class cannot find the records published by ReconcilePublish. Under a different scenario, the same code does find the records. All the Kafka consumer and producer properties are hard-coded in a common ancestor class, to eliminate any possible mismatches.</p>
  <p>If the Reconcile class calls ReconcilePublish and ReconcileSubscribe through their <code>main()</code> methods, everything works as expected.</p>
  <p>If the Reconcile class creates instances of ReconcilePublish and ReconcileSubscribe and then calls their business methods (the same ones called from the <code>main()</code> method), the records are not found.</p>
  <p>In the <code>ReconcileSubscribe::drainTopic(String topicName)</code> method, I print the partition assignment for the Kafka consumer. When things don't work, nothing is printed, i.e. the consumer has not been allocated any partition (but WHY?). This suggests that when the new consumer does not get any allocation, it's because there is already another consumer that owns that partition. But what do I need to do (beyond <code>KafkaConsumer::close()</code>, which I am always doing) to signal that one consumer is detaching form a partition?</p>
  <pre class="lang-java prettyprint-override"><code>// Reconcile
  public void performReconciliation() throws IOException { // THIS WORKS
      ReconcileSubscribe.main(new String[0]); // to drain the topic
      ReconcilePublish.main(new String[0]); // publish 1000 records
      ReconcileSubscribe.main(new String[0]); // drain the topic, count the records
      // RS::main() - ExpectedNumEventsSub: 1000, ActualNumEvents: 1000 // AS EXPECTED
  }
  public void reconcileUsingLocalMethods() { // THIS DOES NOT WORK
      ReconcilePublish pub = new ReconcilePublish();
      ReconcileSubscribe sub = new ReconcileSubscribe();
      int eventsToPublish = 1000;
      int eventsPublished = pub.publishBatch(eventsToPublish, some_topic_name);
      System.out.printf(&quot;rulm() - DesiredNumEventsPub: %d, PublishedEvents: %d%n&quot;, eventsToPublish, eventsPublished);
      int numEvents = sub.drainTopic(some_topic_name);
      System.out.printf(&quot;rulm() - ExpectedNumEventsSub: %d, ActualNumEvents: %d%n&quot;, eventsToPublish, numEvents);
      // rulm() - ExpectedNumEventsSub: 1000, ActualNumEvents: 0 // WRONG. WHY?
  }
  // ReconcilePublish
  public static void main(String[] args) throws IOException {
      ReconcilePublish pub = new ReconcilePublish();
      int eventsToPublish = 1000;
      int eventsPublished = pub.publishBatch(eventsToPublish, some_topic_name);
      System.out.printf(&quot;RP::main() - DesiredNumEventsPub: %d, PublishedEvents: %d%n&quot;, eventsToPublish, eventsPublished);
  }
  // ReconcileSubscribe
  public static void main(String[] args) throws IOException {
      int eventsToPublish = 1000;
      ReconcileSubscribe sub = new ReconcileSubscribe();
      int numEvents = sub.drainTopic(some_topic_name);
      System.out.printf(&quot;RS::main() - ExpectedNumEventsSub: %d, ActualNumEvents: %d%n&quot;, eventsToPublish, numEvents);
  }
  public int drainTopic(String topicName) {
      int numEvents = 0;
      KafkaConsumer&lt;String, String&gt; consumer = createKafkaConsumer();
      consumer.subscribe(singleton(topicName));
      boolean printed = false;
      while (System.currentTimeMillis() - lastEventMillis &lt; 10000L) {
          ConsumerRecords&lt;String, String&gt; records = consumer.poll(ofSeconds(1));
          if (!printed) printed = printAssignedPartitions(consumer); // when ok: &quot;AssignedTopicPartitions:,0&quot; else not reached
          for (ConsumerRecord&lt;String, String&gt; record : records.records(topicName)) {
              lastEventMillis = System.currentTimeMillis();
              numEvents++;
          }
          consumer.commitSync();
      }
      consumer.close();
      return numEvents;
  }
  </code></pre>
  <p>Relevant KafkaConsumer settings:</p>
  <pre><code>props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, &quot;false&quot;);
  props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, &quot;read_committed&quot;);
  props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
  </code></pre>
  <p>Using Kafka 2.7.0, Java 1.8.</p>