public int getKafkaFetcherMinBytes() {
    return _kafkaFetcherMinBytes;
  }
@Test
  public void testGetFetcherMinBytes() {
    // test default
    KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", "", "", "", null, null);
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_FETCHER_MIN_BYTES_DEFAULT,
        config.getKafkaFetcherMinBytes());

    config = getStreamConfig("topic", "host1", "", "", "", "", null);
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_FETCHER_MIN_BYTES_DEFAULT,
        config.getKafkaFetcherMinBytes());

    config = getStreamConfig("topic", "host1", "", "", "", "bad value", null);
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_FETCHER_MIN_BYTES_DEFAULT,
        config.getKafkaFetcherMinBytes());

    // correct config
    config = getStreamConfig("topic", "host1", "", "", "", "100", null);
    Assert.assertEquals(100, config.getKafkaFetcherMinBytes());
  }