public String getAndFormatStatsFor(final String topic, final boolean isError) {
    return format(
        getStatsFor(topic, isError),
        isError ? "last-failed" : "last-message");
  }
@Test
  public void shouldKeepWorkingWhenDuplicateTopicConsumerIsRemoved() {
    final MetricCollectors metricCollectors = new MetricCollectors();

    final ConsumerCollector collector1 = new ConsumerCollector();
    collector1.configure(
        ImmutableMap.of(
            ConsumerConfig.GROUP_ID_CONFIG, "stream-thread-1",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );

    final ConsumerCollector collector2 = new ConsumerCollector();
    collector2.configure(
        ImmutableMap.of(
            ConsumerConfig.GROUP_ID_CONFIG, "stream-thread-2",
            KsqlConfig.KSQL_INTERNAL_METRIC_COLLECTORS_CONFIG, metricCollectors
        )
    );

    final Map<TopicPartition, List<ConsumerRecord<Object, Object>>> records = ImmutableMap.of(
        new TopicPartition(TEST_TOPIC, 1),
        Collections.singletonList(
            new ConsumerRecord<>(
                TEST_TOPIC,
                1,
                1,
                1L,
                TimestampType.CREATE_TIME,
                10,
                10,
                "key",
                "1234567890",
                new RecordHeaders(),
                Optional.empty()
            )
        )
    );
    final ConsumerRecords<Object, Object> consumerRecords = new ConsumerRecords<>(records);


    collector1.onConsume(consumerRecords);
    collector2.onConsume(consumerRecords);

    final String firstPassStats = metricCollectors.getAndFormatStatsFor(TEST_TOPIC, false);

    assertTrue("Missed stats, got:" + firstPassStats, firstPassStats.contains("total-messages:         2"));

    collector2.close();

    collector1.onConsume(consumerRecords);

    final String statsForTopic2 =  metricCollectors.getAndFormatStatsFor(TEST_TOPIC, false);

    assertTrue("Missed stats, got:" + statsForTopic2, statsForTopic2.contains("total-messages:         2"));
  }