static void process(int maxMessages, MessageFormatter formatter, ConsumerWrapper consumer, PrintStream output,
                        boolean rejectMessageOnError, AcknowledgeType acknowledgeType) {
        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);
                consumer.acknowledge(msg, acknowledgeType);
            } catch (Throwable t) {
                if (rejectMessageOnError) {
                    LOG.error("Error processing message, rejecting this message: ", t);
                    consumer.acknowledge(msg, AcknowledgeType.REJECT);
                } else {
                    // Consumer will be closed
                    throw t;
                }
            }
            if (checkErr(output)) {
                // Consumer will be closed
                return;
            }
        }
    }
@Test
    public void shouldStopWhenOutputCheckErrorFails() {
        ConsoleShareConsumer.ConsumerWrapper consumer = mock(ConsoleShareConsumer.ConsumerWrapper.class);
        MessageFormatter formatter = mock(MessageFormatter.class);
        PrintStream printStream = mock(PrintStream.class);

        ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>("foo", 1, 1, new byte[0], new byte[0]);

        when(consumer.receive()).thenReturn(record);
        //Simulate an error on System.out after the first record has been printed
        when(printStream.checkError()).thenReturn(true);

        ConsoleShareConsumer.process(-1, formatter, consumer, printStream, true, AcknowledgeType.ACCEPT);

        verify(formatter).writeTo(any(), eq(printStream));
        verify(consumer).receive();
        verify(printStream).checkError();

        consumer.cleanup();
    }