Batching not working as expected

Hi Team,

I mainly wanted to use the batching feature of Parallel Consumer so started doing a POC around it.

Currently the Kafka topic has 6 partitions and each partition has around 15k messages.

I wanted to consume the data in batches and each batch will be of 10 messages. Below are the code snippets of current parallel consumer configuration. Wanted the data to be consumed in a ordered way.

`ParallelConsumerOptions`

```
final Consumer consumer = new KafkaConsumer<>(appProperties);
final ParallelConsumerOptions options = ParallelConsumerOptions.builder()
.consumer(consumer)
.ordering(PARTITION)
.batchSize(10)
.build();
ParallelStreamProcessor eosStreamProcessor = createEosStreamProcessor(options);
```

`appProperties`
```

# Consumer properties
bootstrap.servers=localhost:9092
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
max.poll.interval.ms=300000
enable.auto.commit=false
auto.offset.reset=earliest

# Application-specific properties
input.topic.name=
```

`Consumer Poll`
```
public void runConsume(final Properties appProperties) {
String topic = appProperties.getProperty("input.topic.name");

LOGGER.info("Subscribing Parallel Consumer to consume from {} topic", topic);
parallelConsumer.subscribe(Collections.singletonList(topic));

LOGGER.info("Polling for records. This method blocks", topic);
parallelConsumer.poll(context -> {
List payload = context.stream().map(this::preparePayload).collect(Collectors.toList());
System.out.println("********* " + payload.size() + " **********");
});
}
```
`preparePayload`

```
private String preparePayload(RecordContext stringStringRecordContext) {

ConsumerRecord consumerRecords = stringStringRecordContext.getConsumerRecord();
int failureCount = stringStringRecordContext.getNumberOfFailedAttempts();
System.out.println("Value: " + consumerRecords.key() + " - Partition: " + consumerRecords.partition() + " Offset: " + consumerRecords.offset());
return msg("{}, {}", consumerRecords, failureCount);
}
```

Now despite setting the batching to 10, the data is being consumed in random size of batches(1,2,3 < 10). Could someone please help me out.

Thanks in advance.

Regards,
Dixit

Start with the supplied ParallelConsumerOptions.builder() configuration, especially batchSize(10), ordering(PARTITION), and the parallelConsumer.poll callback. Reproduce the behavior using the six-partition topic setup and inspect the callback context sizes. Done means establishing whether batches smaller than 10 are expected or identifying the change needed to produce the requested batching behavior.

Written by the indexing model from the issue text.