title: Kafka consumer not getting a single message from partition
body: <p>I just noticed that when I produce a single message into a partition, my consumer is not receiving it. Only after I produce a few more messages into the same partition, the consumer receives them. My <code>fetch.min.bytes</code> is set to 1.</p>
  
  <p>Is there some other config that could affect here?</p>
  
  <p>I have a dedicated consumer for each partition. </p>
  
  <p>Consumer code for the relevant part. My consumer starts several threads for different topics that are defined by the <code>configs['stream']</code>. Uses <a href="https://github.com/mmustala/rdkafka-ruby" rel="nofollow noreferrer">https://github.com/mmustala/rdkafka-ruby</a> which is a fork from original consumer gem. I added a batch consuming method. And a method to shut down the consumer in a managed way.</p>
  
  <pre><code>key = configs['app_key']
  consumer = Rdkafka::Config.new(config(configs)).consumer
  topic = "#{topic_prefix}#{app_env}_#{configs['stream']}"
  consumer.subscribe(topic)
  
  logger.info "#{rand}| Starting consumer for #{key} with topic #{topic}"
  begin
    retry_counter = 0
    retries_started_at = nil
    current_assignment = nil
    partitions = []
    consumer.each_batch(configs['max_messages_per_partition'] || 5, 100, rand) do |messages|
      partitions = messages.collect {|m| m.partition}.uniq.sort
      logger.info "#{rand}| Batch started. Received #{messages.length} messages from partitions #{partitions} for app #{key}"
      current_assignment = consumer.assignment.to_h
      values = messages.collect {|m| JSON.parse(m.payload)}
      skip_commit = false
      begin
        values.each_slice((values.length / ((retry_counter * 2) + 1).to_f).ceil) do |slice|
          logger.info "#{rand}| Sending #{slice.length} messages to lambda"
          result = invoke_lambda(key, slice)
          if result.status_code != 200 || result.function_error
            logger.info "#{rand}| Batch finished with error #{result.function_error}"
            raise LambdaError, result.function_error.to_s
          end
        end
      rescue LambdaError =&gt; e
        logger.warn "#{rand}| #{e}"
        if consumer.running? &amp;&amp; current_assignment == consumer.assignment.to_h
          retry_counter += 1
          retries_started_at ||= Time.now
          if retry_counter &lt;= 5 &amp;&amp; Time.now - retries_started_at &lt; 600
            logger.warn "#{rand}| Retrying from: #{e.cause}, app_key: #{key}"
            Rollbar.warning("Retrying from: #{e.cause}", app_key: key, thread: rand, partitions: partitions.join(', '))
            sleep 5
            retry if consumer.running? &amp;&amp; current_assignment == consumer.assignment.to_h
          else
            raise e # Raise to exit the retry loop so that consumers are rebalanced.
          end
        end
        skip_commit = true
      end
      retry_counter = 0
      retries_started_at = nil
      if skip_commit
        logger.info "#{rand}| Commit skipped"
      else
        consumer.commit
        logger.info "#{rand}| Batch finished"
      end
    end
    consumer.close
    logger.info "#{rand}| Stopped #{key}"
  rescue Rdkafka::RdkafkaError =&gt; e
    logger.warn "#{rand}| #{e}"
    logger.info "#{rand}| assignment: #{consumer.assignment.to_h}"
    if e.to_s.index('No offset stored')
      retry
    else
      raise e
    end
  end
  </code></pre>
  
  <p>config</p>
  
  <pre><code>def config(app_config)
    {
        "bootstrap.servers": brokers,
        "group.id": app_configs['app_key'],
        "enable.auto.commit": false,
        "enable.partition.eof": false,
        "log.connection.close": false,
        "session.timeout.ms": 30*1000,
        "fetch.message.max.bytes": ['sources'].include?(app_configs['stream']) ? 102400 : 10240,
        "queued.max.messages.kbytes": ['sources'].include?(app_configs['stream']) ? 250 : 25,
        "queued.min.messages": (app_configs['max_messages_per_partition'] || 5) * 10,
        "fetch.min.bytes": 1,
        "partition.assignment.strategy": 'roundrobin'
    }
  end
  </code></pre>
  
  <p>Producer code uses <a href="https://github.com/zendesk/ruby-kafka" rel="nofollow noreferrer">https://github.com/zendesk/ruby-kafka</a></p>
  
  <pre><code>def to_kafka(stream_name, data, batch_size)
    stream_name_with_env = "#{Rails.env}_#{stream_name}"
    topic = [Rails.application.secrets.kafka_topic_prefix, stream_name_with_env].compact.join
    partitions_count = KAFKA.partitions_for(topic)
    Rails.logger.info "Partition count for #{topic}: #{partitions_count}"
    if @job.active? &amp;&amp; @job.partition.blank?
      @job.connect_to_partition
    end
    partition = @job.partition&amp;.number.to_i % partitions_count
    producer = KAFKA.producer 
    if data.is_a?(Array)
      data.each_slice(batch_size) do |slice|
        producer.produce(JSON.generate(slice), topic: topic, partition: partition)
      end
    else
      producer.produce(JSON.generate(data), topic: topic, partition: partition)
    end
    producer.deliver_messages
    Rails.logger.info "records sent to topic #{topic} partition #{partition}"
    producer.shutdown
  end
  </code></pre>
  
  <p>UPDATE: It looks like the number of messages is irrelevant. I just produced over 100 messages into one partition and the consumer has not yet started to consume those. </p>
  
  <p>UPDATE2: It didn't start consuming the messages during the night. But when I produced a new set of messages into the same partition this morning, it woke up and started to consume the new messages I just produced. It skipped over the messages produced last night.</p>