// Path: hadoop-mini-clusters-kafka/src/main/java/com/github/sakserv/minicluster/systemtime/LocalSystemTime.java
// public class LocalSystemTime implements Time {
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
//             // no stress
//         }
//     }
// 
//     @Override
//     public long hiResClockMs() {
//         return System.currentTimeMillis();
//     }
// 
// }
// 
// Path: hadoop-mini-clusters-common/src/main/java/com/github/sakserv/minicluster/util/FileUtils.java
// public final class FileUtils {
// 
//     // Logger
//     private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(FileUtils.class);
// 
//     public static void deleteFolder(String directory, boolean quietly) {
//         try {
//             Path directoryPath = Paths.get(directory).toAbsolutePath();
//             if (!quietly) {
//                 LOG.info("FILEUTILS: Deleting contents of directory: {}",
//                         directoryPath.toAbsolutePath().toString());
//             }
//             Files.walkFileTree(directoryPath, new SimpleFileVisitor<Path>() {
//                 @Override
//                 public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
//                         throws IOException {
//                     Files.delete(file);
//                     if (!quietly) {
//                         LOG.info("Removing file: {}", file.toAbsolutePath().toString());
//                     }
//                     return FileVisitResult.CONTINUE;
//                 }
// 
//                 @Override
//                 public FileVisitResult postVisitDirectory(Path dir, IOException exc)
//                         throws IOException {
//                     Files.delete(dir);
//                     if (!quietly) {
//                         LOG.info("Removing directory: {}", dir.toAbsolutePath().toString());
//                     }
//                     return FileVisitResult.CONTINUE;
//                 }
//             });
//         } catch (IOException e) {
//             LOG.error("FILEUTILS: Unable to remove {}", directory);
//         }
//     }
// 
//     public static void deleteFolder(String directory) {
//         deleteFolder(directory, false);
//     }
// 
//     @Override
//     public String toString() {
//         return "FileUtils";
//     }
// }


import scala.collection.Seq;
import java.lang.reflect.Constructor;
import java.util.Properties;
import kafka.metrics.KafkaMetricsReporter;
import kafka.metrics.KafkaMetricsReporter$;
import kafka.utils.VerifiableProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.sakserv.minicluster.MiniCluster;
import com.github.sakserv.minicluster.systemtime.LocalSystemTime;
import com.github.sakserv.minicluster.util.FileUtils;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import scala.Option;

        // The Kafka API for KafkaServer has changed multiple times.
        // Using reflection to call the version specific constructor
        // with the correct number of arguments.

        // We only expect a single constructor, throw an Exception if there is not exactly 1
        Class<KafkaServer> kafkaServerClazz = KafkaServer.class;
        Constructor[] kafkaServerConstructors =  kafkaServerClazz.getConstructors();
        if(kafkaServerConstructors.length != 1) {
            throw new Exception("kafka.server.KafkaServer has more than one constructor, expected only 1");
        }

        // We only expect 2, 3 and 4 argument constructors, throw an Exception if not
        Constructor kafkaServerConstructor = kafkaServerConstructors[0];

        // Kafka 2.9.2 0.8.2 (HDP 2.3.0 and HDP 2.3.2)
        if (kafkaServerConstructor.getParameterTypes().length == 2) {
            kafkaServer = (KafkaServer) kafkaServerConstructor.newInstance(kafkaConfig, new LocalSystemTime());

        // Kafka 2.10 0.9.0 (HDP 2.3.4), pass in the threadPrefixName
        } else if (kafkaServerConstructor.getParameterTypes().length == 3) {
            Option<String> threadPrefixName = Option.apply("kafka-mini-cluster");
            kafkaServer = (KafkaServer) kafkaServerConstructor.newInstance(kafkaConfig, new LocalSystemTime(), threadPrefixName);

        // Kafka 2.10.1
        } else if (kafkaServerConstructor.getParameterTypes().length == 4) {
            //val config: KafkaConfig, time: Time = Time.SYSTEM, threadNamePrefix: Option[String] = None, kafkaMetricsReporters: Seq[KafkaMetricsReporter] = List()
            Option<String> threadPrefixName = Option.apply("kafka-mini-cluster");
            //scala.collection.immutable.List<Object> kafkaMetricsReporters = scala.collection.immutable.List.empty();
            Seq<KafkaMetricsReporter> reporters = KafkaMetricsReporter$.MODULE$.startReporters(new VerifiableProperties(new Properties()));
// Path: hadoop-mini-clusters-kafka/src/main/java/com/github/sakserv/minicluster/systemtime/LocalSystemTime.java
// public class LocalSystemTime implements Time {
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
//             // no stress
//         }
//     }
// 
//     @Override
//     public long hiResClockMs() {
//         return System.currentTimeMillis();
//     }
// 
// }
// 
// Path: hadoop-mini-clusters-common/src/main/java/com/github/sakserv/minicluster/util/FileUtils.java
// public final class FileUtils {
// 
//     // Logger
//     private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(FileUtils.class);
// 
//     public static void deleteFolder(String directory, boolean quietly) {
//         try {
//             Path directoryPath = Paths.get(directory).toAbsolutePath();
//             if (!quietly) {
//                 LOG.info("FILEUTILS: Deleting contents of directory: {}",
//                         directoryPath.toAbsolutePath().toString());
//             }
//             Files.walkFileTree(directoryPath, new SimpleFileVisitor<Path>() {
//                 @Override
//                 public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
//                         throws IOException {
//                     Files.delete(file);
//                     if (!quietly) {
//                         LOG.info("Removing file: {}", file.toAbsolutePath().toString());
//                     }
//                     return FileVisitResult.CONTINUE;
//                 }
// 
//                 @Override
//                 public FileVisitResult postVisitDirectory(Path dir, IOException exc)
//                         throws IOException {
//                     Files.delete(dir);
//                     if (!quietly) {
//                         LOG.info("Removing directory: {}", dir.toAbsolutePath().toString());
//                     }
//                     return FileVisitResult.CONTINUE;
//                 }
//             });
//         } catch (IOException e) {
//             LOG.error("FILEUTILS: Unable to remove {}", directory);
//         }
//     }
// 
//     public static void deleteFolder(String directory) {
//         deleteFolder(directory, false);
//     }
// 
//     @Override
//     public String toString() {
//         return "FileUtils";
//     }
// }


// Path: hadoop-mini-clusters-kafka/src/main/java/com/github/sakserv/minicluster/impl/KafkaLocalBroker.java
import scala.collection.Seq;
import java.lang.reflect.Constructor;
import java.util.Properties;
import kafka.metrics.KafkaMetricsReporter;
import kafka.metrics.KafkaMetricsReporter$;
import kafka.utils.VerifiableProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.sakserv.minicluster.MiniCluster;
import com.github.sakserv.minicluster.systemtime.LocalSystemTime;
import com.github.sakserv.minicluster.util.FileUtils;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import scala.Option;


        // The Kafka API for KafkaServer has changed multiple times.
        // Using reflection to call the version specific constructor
        // with the correct number of arguments.

        // We only expect a single constructor, throw an Exception if there is not exactly 1
        Class<KafkaServer> kafkaServerClazz = KafkaServer.class;
        Constructor[] kafkaServerConstructors =  kafkaServerClazz.getConstructors();
        if(kafkaServerConstructors.length != 1) {
            throw new Exception("kafka.server.KafkaServer has more than one constructor, expected only 1");
        }

        // We only expect 2, 3 and 4 argument constructors, throw an Exception if not
        Constructor kafkaServerConstructor = kafkaServerConstructors[0];

        // Kafka 2.9.2 0.8.2 (HDP 2.3.0 and HDP 2.3.2)
        if (kafkaServerConstructor.getParameterTypes().length == 2) {
            kafkaServer = (KafkaServer) kafkaServerConstructor.newInstance(kafkaConfig, new LocalSystemTime());

        // Kafka 2.10 0.9.0 (HDP 2.3.4), pass in the threadPrefixName
        } else if (kafkaServerConstructor.getParameterTypes().length == 3) {
            Option<String> threadPrefixName = Option.apply("kafka-mini-cluster");
            kafkaServer = (KafkaServer) kafkaServerConstructor.newInstance(kafkaConfig, new LocalSystemTime(), threadPrefixName);

        // Kafka 2.10.1
        } else if (kafkaServerConstructor.getParameterTypes().length == 4) {
            //val config: KafkaConfig, time: Time = Time.SYSTEM, threadNamePrefix: Option[String] = None, kafkaMetricsReporters: Seq[KafkaMetricsReporter] = List()
            Option<String> threadPrefixName = Option.apply("kafka-mini-cluster");
            //scala.collection.immutable.List<Object> kafkaMetricsReporters = scala.collection.immutable.List.empty();
            Seq<KafkaMetricsReporter> reporters = KafkaMetricsReporter$.MODULE$.startReporters(new VerifiableProperties(new Properties()));
            kafkaServer = (KafkaServer) kafkaServerConstructor.newInstance(kafkaConfig, new LocalSystemTime(), threadPrefixName, reporters);