public int getKafkaSocketTimeout() {
    return _kafkaSocketTimeout;
  }
@Test
  public void testGetKafkaSocketTimeout() {
    // test default
    KafkaPartitionLevelStreamConfig config = getStreamConfig("topic", "host1", "", null);
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_SOCKET_TIMEOUT_DEFAULT,
        config.getKafkaSocketTimeout());

    config = getStreamConfig("topic", "host1", "", "");
    Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_SOCKET_TIMEOUT_DEFAULT,
        config.getKafkaSocketTimeout());

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

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