// 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;
    public static boolean seekToEnd(KafkaConsumer<?, ?> consumer, String topic) {
        boolean result = false;
        synchronized (consumer) {
            Set<TopicPartition> topicParts = consumer.assignment();
            if (topicParts != null) {
                for (TopicPartition tp : topicParts) {
                    if (StringUtils.equals(topic, tp.topic())) {
                        consumer.seekToEnd(Arrays.asList(tp));
                        /* we want to seek as soon as possible  since seekToEnd evaluates lazily,
                         * invoke position() so that seeking will be committed.
                         */
                        consumer.position(tp);
                        result = true;
                    }
                }
                if (result) {
                    consumer.commitSync();
                }
            }
        }
        return result;
    }

    /**
     * Create a new {@link KafkaProducer} instance, with default configurations.
     *
     * @param type
     * @param bootstrapServers
     * @return
     */
// 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;

    public static boolean seekToEnd(KafkaConsumer<?, ?> consumer, String topic) {
        boolean result = false;
        synchronized (consumer) {
            Set<TopicPartition> topicParts = consumer.assignment();
            if (topicParts != null) {
                for (TopicPartition tp : topicParts) {
                    if (StringUtils.equals(topic, tp.topic())) {
                        consumer.seekToEnd(Arrays.asList(tp));
                        /* we want to seek as soon as possible  since seekToEnd evaluates lazily,
                         * invoke position() so that seeking will be committed.
                         */
                        consumer.position(tp);
                        result = true;
                    }
                }
                if (result) {
                    consumer.commitSync();
                }
            }
        }
        return result;
    }

    /**
     * Create a new {@link KafkaProducer} instance, with default configurations.
     *
     * @param type
     * @param bootstrapServers
     * @return
     */
    public static KafkaProducer<String, byte[]> createKafkaProducer(ProducerType type, String bootstrapServers) {