I have also been looking at this functionality to support long polling.

I think the behaviour that you describe above is as per "spec" in that to send a larger message through Kafka Rest would have to exceed the max message you have specified. The tests certainly seem to back that up. Would be great if one of the core maintainers comment on this...

Having said that, its rather inconvenient as I want the same behaviour that you describe - a sort of byte threshold rather than max.

Finally, when investigating this I have discovered a bug. The following code is the loop that handles reading messages from kafka. The loop correctly breaks when the max bytes is exceeded.

        while (iter.hasNext()) {
          MessageAndMetadata<KafkaK, KafkaV> msg = iter.peek();
          ConsumerRecordAndSize<ClientK, ClientV> recordAndSize = parent.createConsumerRecord(msg);
          long roughMsgSize = recordAndSize.getSize();
          if (bytesConsumed + roughMsgSize > maxResponseBytes) {
            break; // Loop breaks here due to max bytes being exceeded
          }

          iter.next();
          messages.add(recordAndSize.getRecord());
          bytesConsumed += roughMsgSize;
          // Updating the consumed offsets isn't done until we're actually going to return the
          // data since we may encounter an error during a subsequent read, in which case we'll
          // have to defer returning the data so we can return an HTTP error instead
        }

But then later on the check to determine if the processing of messages has finished the check fails to take into account the roughMsgSize and so will not call finish. The only way finish can be called is for the timeout to elapse. So, for a long poll example of 30s the messages won't be picked up until that 30s has elapsed...

   if (elapsed >= requestTimeoutMs || bytesConsumed >= maxResponseBytes) {
        finish();
      }

I'll be submitting a patch to fix this particular issue in the next few days.

Thanks,
Alan

#155