static void process(int maxMessages, MessageFormatter formatter, ConsumerWrapper consumer, PrintStream output, boolean skipMessageOnError) {
        while (messageCount < maxMessages || maxMessages == -1) {
            ConsumerRecord<byte[], byte[]> msg;
            try {
                msg = consumer.receive();
            } catch (WakeupException we) {
                LOG.trace("Caught WakeupException because consumer is shutdown, ignore and terminate.");
                // Consumer will be closed
                return;
            } catch (Throwable t) {
                LOG.error("Error processing message, terminating consumer process: ", t);
                // Consumer will be closed
                return;
            }
            messageCount += 1;
            try {
                formatter.writeTo(new ConsumerRecord<>(msg.topic(), msg.partition(), msg.offset(), msg.timestamp(), msg.timestampType(),
                    0, 0, msg.key(), msg.value(), msg.headers(), Optional.empty()), output);
            } catch (Throwable t) {
                if (skipMessageOnError) {
                    LOG.error("Error processing message, skipping this message: ", t);
                } else {
                    // Consumer will be closed
                    throw t;
                }
            }
            if (checkErr(output)) {
                // Consumer will be closed
                return;
            }
        }
    }
@Test
    public void shouldResetUnConsumedOffsetsBeforeExit() throws IOException {
        String topic = "test";
        int maxMessages = 123;
        int totalMessages = 700;
        long startOffset = 0L;

        MockConsumer<byte[], byte[]> mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        TopicPartition tp1 = new TopicPartition(topic, 0);
        TopicPartition tp2 = new TopicPartition(topic, 1);

        String[] args = new String[]{
            "--bootstrap-server", "localhost:9092",
            "--topic", topic,
            "--timeout-ms", "1000"
        };

        ConsoleConsumer.ConsumerWrapper consumer = new ConsoleConsumer.ConsumerWrapper(
            new ConsoleConsumerOptions(args),
            mockConsumer
        );

        mockConsumer.rebalance(Arrays.asList(tp1, tp2));
        Map<TopicPartition, Long> offsets = new HashMap<>();
        offsets.put(tp1, startOffset);
        offsets.put(tp2, startOffset);
        mockConsumer.updateBeginningOffsets(offsets);

        for (int i = 0; i < totalMessages; i++) {
            // add all records, each partition should have half of `totalMessages`
            mockConsumer.addRecord(new ConsumerRecord<>(topic, i % 2, i / 2, "key".getBytes(), "value".getBytes()));
        }

        MessageFormatter formatter = mock(MessageFormatter.class);

        ConsoleConsumer.process(maxMessages, formatter, consumer, System.out, false);
        assertEquals(totalMessages, mockConsumer.position(tp1) + mockConsumer.position(tp2));

        consumer.resetUnconsumedOffsets();
        assertEquals(maxMessages, mockConsumer.position(tp1) + mockConsumer.position(tp2));

        verify(formatter, times(maxMessages)).writeTo(any(), any());
        consumer.cleanup();
    }