// Path: src/main/java/com/github/ddth/kafka/KafkaClient.java
// public enum ProducerType {NO_ACK, LEADER_ACK, ALL_ACKS;
//     public final static ProducerType[] ALL_TYPES = { NO_ACK, LEADER_ACK, ALL_ACKS };}
// 
// Path: src/main/java/com/github/ddth/kafka/KafkaTopicPartitionOffset.java
// public class KafkaTopicPartitionOffset {
//     public final String topic;
//     public final int partition;
//     public final long offset;
// 
//     public KafkaTopicPartitionOffset(String topic, int partition, long offset) {
//         this.topic = topic;
//         this.partition = partition;
//         this.offset = offset;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public String toString() {
//         ToStringBuilder tsb = new ToStringBuilder(this);
//         tsb.append("topic", topic).append("partition", partition).append("offset", offset);
//         return tsb.build();
//     }
// }


import com.github.ddth.kafka.KafkaClient.ProducerType;
import com.github.ddth.kafka.KafkaTopicPartitionOffset;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Arrays;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
package com.github.ddth.kafka.internal;



/**
 * Helper utility class.
 *
 * @author Thanh Ba Nguyen <bnguyen2k@gmail.com>
 * @since 1.2.0
 */
public class KafkaHelper {
    /**
     * Helper method to create a {@link ExecutorService} instance (internal use).
     *
     * @param es
     * @return
     */
    public static ExecutorService createExecutorServiceIfNull(ExecutorService es) {
        if (es == null) {
            int numThreads = Math.min(Math.max(Runtime.getRuntime().availableProcessors(), 1), 4);
            es = Executors.newFixedThreadPool(numThreads);
        }
        return es;
    }

    /**
     * Helper method to shutdown a {@link ExecutorService} instance (internal use).
     *
     * @param es
     */
    public static void destroyExecutorService(ExecutorService es) {
        if (es != null) {
            es.shutdownNow();
        }
    }

    /**
     * Seek to a specified offset.
     *
     * @param consumer
     * @param tpo
     * @return {@code true} if the consumer has subscribed to the specified
     * topic/partition, {@code false} otherwise.
     * @since 1.3.2
     */
// Path: src/main/java/com/github/ddth/kafka/KafkaClient.java
// public enum ProducerType {NO_ACK, LEADER_ACK, ALL_ACKS;
//     public final static ProducerType[] ALL_TYPES = { NO_ACK, LEADER_ACK, ALL_ACKS };}
// 
// Path: src/main/java/com/github/ddth/kafka/KafkaTopicPartitionOffset.java
// public class KafkaTopicPartitionOffset {
//     public final String topic;
//     public final int partition;
//     public final long offset;
// 
//     public KafkaTopicPartitionOffset(String topic, int partition, long offset) {
//         this.topic = topic;
//         this.partition = partition;
//         this.offset = offset;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public String toString() {
//         ToStringBuilder tsb = new ToStringBuilder(this);
//         tsb.append("topic", topic).append("partition", partition).append("offset", offset);
//         return tsb.build();
//     }
// }


// Path: src/main/java/com/github/ddth/kafka/internal/KafkaHelper.java
import com.github.ddth.kafka.KafkaClient.ProducerType;
import com.github.ddth.kafka.KafkaTopicPartitionOffset;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Arrays;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

package com.github.ddth.kafka.internal;



/**
 * Helper utility class.
 *
 * @author Thanh Ba Nguyen <bnguyen2k@gmail.com>
 * @since 1.2.0
 */
public class KafkaHelper {
    /**
     * Helper method to create a {@link ExecutorService} instance (internal use).
     *
     * @param es
     * @return
     */
    public static ExecutorService createExecutorServiceIfNull(ExecutorService es) {
        if (es == null) {
            int numThreads = Math.min(Math.max(Runtime.getRuntime().availableProcessors(), 1), 4);
            es = Executors.newFixedThreadPool(numThreads);
        }
        return es;
    }

    /**
     * Helper method to shutdown a {@link ExecutorService} instance (internal use).
     *
     * @param es
     */
    public static void destroyExecutorService(ExecutorService es) {
        if (es != null) {
            es.shutdownNow();
        }
    }

    /**
     * Seek to a specified offset.
     *
     * @param consumer
     * @param tpo
     * @return {@code true} if the consumer has subscribed to the specified
     * topic/partition, {@code false} otherwise.
     * @since 1.3.2
     */
    public static boolean seek(KafkaConsumer<?, ?> consumer, KafkaTopicPartitionOffset tpo) {