static Properties producerProperties(Configuration configuration) {
    final String writeSemanticValue = configuration.get(KafkaTableProperties.WRITE_SEMANTIC_PROPERTY.getName());
    final KafkaOutputFormat.WriteSemantic writeSemantic = KafkaOutputFormat.WriteSemantic.valueOf(writeSemanticValue);
    final Properties properties = new Properties();
    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());
    }
    properties.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerEndPoint);
    //case Kerberos is On
    if (UserGroupInformation.isSecurityEnabled()) {
      addKerberosJaasConf(configuration, properties);
    }

    // user can always override stuff
    properties.putAll(extractExtraProperties(configuration, PRODUCER_CONFIGURATION_PREFIX));
    setupKafkaSslProperties(configuration, properties);

    String taskId = configuration.get("mapred.task.id", null);
    properties.setProperty(CommonClientConfigs.CLIENT_ID_CONFIG,
        taskId == null ? "random_" + UUID.randomUUID().toString() : taskId);
    switch (writeSemantic) {
    case AT_LEAST_ONCE:
      properties.setProperty(ProducerConfig.RETRIES_CONFIG, String.valueOf(Integer.MAX_VALUE));
      //The number of acknowledgments the producer requires the leader to have received before considering a request as
      //complete. Here all means from all replicas.
      properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
      break;
    case EXACTLY_ONCE:
      // Assuming that TaskId is ReducerId_attemptId. need the Reducer ID to fence out zombie kafka producers.
      String reducerId = getTaskId(configuration);
      //The number of acknowledgments the producer requires the leader to have received before considering a request as
      // complete, all means from all replicas.
      properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
      properties.setProperty(ProducerConfig.RETRIES_CONFIG, String.valueOf(Integer.MAX_VALUE));
      properties.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, reducerId);
      //Producer set to be IDEMPOTENT eg ensure that send() retries are idempotent.
      properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
      break;
    default:
      throw new IllegalArgumentException("Unknown Semantic " + writeSemantic);
    }
    return properties;
  }
Copies Kafka SSL properties from source configuration to target property map.
<p>
It only copies SSL properties that are present in the source and not present in the target. It is useful to
propagate global configurations to the Kafka client but also account for use-cases where table properties are not
using the Hive specific prefixes ({@link #CONSUMER_CONFIGURATION_PREFIX}, {@link #PRODUCER_CONFIGURATION_PREFIX}).
</p>
@param source the configuration from which we will get the properties 
@param target the property map to which we will set the properties
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaUtils.java