// Path: src/main/java/com/neverwinterdp/sparkngin/NullDevMessageForwarder.java
// public class NullDevMessageForwarder implements MessageForwarder {
//   private boolean dump = false ;
//   private int count ;
//   private Throwable error ;
//   
//   
//   public NullDevMessageForwarder() {
//   }
//   
//   public NullDevMessageForwarder(Map<String, String> props) {
//     dump = MapUtil.getBool(props, "forwarder.nulldev.dump", false) ;
//   }
//   
//   public boolean hasError() { return error != null ; }
//   
//   public void setError(Throwable error) { this.error = error ; }
//   
//   public boolean reconnect() { 
//     error = null ;
//     return true ; 
//   }
//   
//   public int getProcessCount() { return count ; }
//   
//   public void forward(Message message) {
//     count++ ;
//     dump(message) ;
//   }
//   
//   public void dump(Message message) {
//     if(!dump) return ;
//     System.out.println(JSONSerializer.INSTANCE.toString(message.getHeader()));
//     try {
//       Class<?> type = Class.forName(message.getData().getType()) ;
//       Object object = message.getData().getDataAs(type) ;
//       System.out.println(JSONSerializer.INSTANCE.toString(object));
//     } catch (ClassNotFoundException e) {
//       e.printStackTrace();
//     }
//   }
//   
//   public void close() {
//   }
// }


import com.neverwinterdp.server.Server;
import com.neverwinterdp.server.shell.Shell;
import com.neverwinterdp.sparkngin.NullDevMessageForwarder;
import com.neverwinterdp.util.FileUtil;

    String kafkaReplication = kafkaServer.length >= 2 ? "2" : "1" ;
    for(int i  = 0; i < kafkaServer.length; i++) {
      int id = i + 1;
      shell.execute(
          "module install "+ 
          "  --member-name kafka" + id +
          "  --autostart" +
          "  --module Kafka" +
          "  -Pmodule.data.drop=true" +
          "  -Pkafka:broker.id=" + id +
          "  -Pkafka:port=" + (9092 + i) +
          "  -Pkafka:zookeeper.connect=127.0.0.1:2181" +
          "  -Pkafka:default.replication.factor=" + kafkaReplication +
          "  -Pkafka:controller.socket.timeout.ms=90000" +
          "  -Pkafka:controlled.shutdown.enable=true" + 
          "  -Pkafka:controlled.shutdown.max.retries=3" +
          "  -Pkafka:controlled.shutdown.retry.backoff.ms=60000"
      ) ;
    }
    shell.execute(
        "module install " + 
        "  -Pmodule.data.drop=true" +
        "  -Pkafka:zookeeper.connect=127.0.0.1:2181" +
        "  --member-role generic --autostart --module KafkaConsumer"
    ) ;
    
    shell.execute(
        "module install" +
        "  -Pmodule.data.drop=true" +
        "  -Psparkngin:http-listen-port=7080" +

// Path: src/main/java/com/neverwinterdp/sparkngin/NullDevMessageForwarder.java
// public class NullDevMessageForwarder implements MessageForwarder {
//   private boolean dump = false ;
//   private int count ;
//   private Throwable error ;
//   
//   
//   public NullDevMessageForwarder() {
//   }
//   
//   public NullDevMessageForwarder(Map<String, String> props) {
//     dump = MapUtil.getBool(props, "forwarder.nulldev.dump", false) ;
//   }
//   
//   public boolean hasError() { return error != null ; }
//   
//   public void setError(Throwable error) { this.error = error ; }
//   
//   public boolean reconnect() { 
//     error = null ;
//     return true ; 
//   }
//   
//   public int getProcessCount() { return count ; }
//   
//   public void forward(Message message) {
//     count++ ;
//     dump(message) ;
//   }
//   
//   public void dump(Message message) {
//     if(!dump) return ;
//     System.out.println(JSONSerializer.INSTANCE.toString(message.getHeader()));
//     try {
//       Class<?> type = Class.forName(message.getData().getType()) ;
//       Object object = message.getData().getDataAs(type) ;
//       System.out.println(JSONSerializer.INSTANCE.toString(object));
//     } catch (ClassNotFoundException e) {
//       e.printStackTrace();
//     }
//   }
//   
//   public void close() {
//   }
// }


// Path: src/test/java/com/neverwinterdp/sparkngin/http/SparknginClusterBuilder.java
import com.neverwinterdp.server.Server;
import com.neverwinterdp.server.shell.Shell;
import com.neverwinterdp.sparkngin.NullDevMessageForwarder;
import com.neverwinterdp.util.FileUtil;

    String kafkaReplication = kafkaServer.length >= 2 ? "2" : "1" ;
    for(int i  = 0; i < kafkaServer.length; i++) {
      int id = i + 1;
      shell.execute(
          "module install "+ 
          "  --member-name kafka" + id +
          "  --autostart" +
          "  --module Kafka" +
          "  -Pmodule.data.drop=true" +
          "  -Pkafka:broker.id=" + id +
          "  -Pkafka:port=" + (9092 + i) +
          "  -Pkafka:zookeeper.connect=127.0.0.1:2181" +
          "  -Pkafka:default.replication.factor=" + kafkaReplication +
          "  -Pkafka:controller.socket.timeout.ms=90000" +
          "  -Pkafka:controlled.shutdown.enable=true" + 
          "  -Pkafka:controlled.shutdown.max.retries=3" +
          "  -Pkafka:controlled.shutdown.retry.backoff.ms=60000"
      ) ;
    }
    shell.execute(
        "module install " + 
        "  -Pmodule.data.drop=true" +
        "  -Pkafka:zookeeper.connect=127.0.0.1:2181" +
        "  --member-role generic --autostart --module KafkaConsumer"
    ) ;
    
    shell.execute(
        "module install" +
        "  -Pmodule.data.drop=true" +
        "  -Psparkngin:http-listen-port=7080" +

        "  -Psparkngin:forwarder-class=" + NullDevMessageForwarder.class.getName() +
