public int getKafkaBufferSize() {
    return _kafkaBufferSize;
  }
@Test
  public void testGetKafkaBufferSize() {
    // test default
    KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", null, "");
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
        config.getKafkaBufferSize());

    config = getStreamConfig("topic", "host1", "", "");
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
        config.getKafkaBufferSize());

    config = getStreamConfig("topic", "host1", "bad value", "");
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT,
        config.getKafkaBufferSize());

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