static Properties consumerProperties(Configuration configuration) {
    final Properties props = new Properties();
    // we are managing the commit offset
    props.setProperty(CommonClientConfigs.CLIENT_ID_CONFIG, Utilities.getTaskId(configuration));
    props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    // we are seeking in the stream so no reset
    props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
    String brokerEndPoint = configuration.get(KafkaTableProperties.HIVE_KAFKA_BOOTSTRAP_SERVERS.getName());
    if (brokerEndPoint == null || brokerEndPoint.isEmpty()) {
      throw new IllegalArgumentException("Kafka Broker End Point is missing Please set Config "
          + KafkaTableProperties.HIVE_KAFKA_BOOTSTRAP_SERVERS.getName());
    }
    props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerEndPoint);
    props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
    props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
    //case Kerberos is On
    if (UserGroupInformation.isSecurityEnabled()) {
      addKerberosJaasConf(configuration, props);
    }

    // user can always override stuff, but SSL properties are derived from configuration, because they require local
    //   files. These need to modified afterwards. This works because these properties use the standard consumer prefix.
    props.putAll(extractExtraProperties(configuration, CONSUMER_CONFIGURATION_PREFIX));
    setupKafkaSslProperties(configuration, props);

    return props;
  }
@param configuration Job configs

@return default consumer properties
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaUtils.java