package com.mesosphere.sdk.kafka.scheduler;

import com.google.protobuf.TextFormat;
import com.mesosphere.sdk.kafka.api.BrokerController;
import com.mesosphere.sdk.scheduler.SchedulerDriverFactory;
import com.mesosphere.sdk.specification.ServiceSpec;
import com.mesosphere.sdk.curator.CuratorUtils;
import com.mesosphere.sdk.scheduler.DefaultScheduler;
import com.mesosphere.sdk.specification.DefaultService;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import com.mesosphere.sdk.specification.yaml.RawServiceSpec;
import com.mesosphere.sdk.specification.yaml.YAMLServiceSpecFactory;
import org.apache.mesos.Protos;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.*;

/**
 * Kafka Service.
 */
public class KafkaService extends  DefaultService {
    protected static final int FAILOVER_TIMEOUT_SEC = 2 * 7 * 24 * 60 * 60;
    protected static final int LOCK_ATTEMPTS = 3;
    protected static final String USER = "root";
    protected static final String LOCK_PATH = "lock";

    protected static final Logger LOGGER = LoggerFactory.getLogger(KafkaService.class);

    public KafkaService(File pathToYamlSpecification) throws Exception {

        //TODO(Mehmet):  call DefaultService(DefaultScheduler.Builder schedulerBuilder) and override register
        super();

        RawServiceSpec rawServiceSpec = YAMLServiceSpecFactory.generateRawSpecFromYAML(pathToYamlSpecification);

        DefaultScheduler.Builder schedulerBuilder =
                DefaultScheduler.newBuilder(YAMLServiceSpecFactory.generateServiceSpec(rawServiceSpec));

        schedulerBuilder.setPlansFrom(rawServiceSpec);

        CuratorFramework curatorClient = CuratorFrameworkFactory.newClient(
                schedulerBuilder.getServiceSpec().getZookeeperConnection(), CuratorUtils.getDefaultRetry());
        curatorClient.start();

        InterProcessMutex curatorMutex = super.lock(curatorClient, schedulerBuilder.getServiceSpec().getName(),
                LOCK_PATH, LOCK_ATTEMPTS);
        try {
            registerAndRun(schedulerBuilder);
        } finally {
            super.unlock(curatorMutex);
            curatorClient.close();
        }

    }

    public void registerAndRun(DefaultScheduler.Builder schedulerBuilder){

        DefaultScheduler scheduler = schedulerBuilder.build();

        ServiceSpec serviceSpec = schedulerBuilder.getServiceSpec();

        String zookeeperConnection = serviceSpec.getZookeeperConnection();

        Protos.FrameworkInfo frameworkInfo =
                super.getFrameworkInfo(serviceSpec, schedulerBuilder.getStateStore(), USER, FAILOVER_TIMEOUT_SEC);

        Collection<Object> jsonResources = new ArrayList<>();

        LOGGER.info("Registering framework: {}", TextFormat.shortDebugString(frameworkInfo));
        new SchedulerDriverFactory().create(scheduler,
                    frameworkInfo,
                    String.format("zk://%s/mesos", zookeeperConnection)).run();

        jsonResources.add(new BrokerController(System.getenv("KAFKA_ZOOKEEPER_URI")));

        startApiServer(scheduler,  serviceSpec.getApiPort(), jsonResources);

    }
}

@@ -1320,119 +1320,8 @@
 on %7B
-%0A%0A        //TODO(Mehmet):  call DefaultService(DefaultScheduler.Builder schedulerBuilder) and override register
 %0A   
@@ -2196,25 +2196,24 @@
 ;%0A        %7D%0A
-%0A
     %7D%0A%0A    p
@@ -2286,73 +2286,8 @@
 r)%7B%0A
-%0A        DefaultScheduler scheduler = schedulerBuilder.build();%0A%0A
     
@@ -2350,16 +2350,79 @@
 eSpec();
+%0A        DefaultScheduler scheduler = schedulerBuilder.build();
 %0A%0A      
@@ -2485,24 +2485,323 @@
 onnection();
+%0A        Collection%3CObject%3E jsonResources = new ArrayList%3C%3E();%0A%0A        //TODO: do error handling if env var does not exist%0A        jsonResources.add(new BrokerController(System.getenv(%22TASKCFG_ALL_KAFKA_ZOOKEEPER_URI%22)));%0A        startApiServer(scheduler,  serviceSpec.getApiPort(), jsonResources);
 %0A%0A        Pr
@@ -2955,71 +2955,8 @@
 );%0A%0A
-        Collection%3CObject%3E jsonResources = new ArrayList%3C%3E();%0A%0A
     
@@ -3135,16 +3135,16 @@
 rkInfo,%0A
+
         
@@ -3219,175 +3219,8 @@
 ();%0A
-%0A        jsonResources.add(new BrokerController(System.getenv(%22KAFKA_ZOOKEEPER_URI%22)));%0A%0A        startApiServer(scheduler,  serviceSpec.getApiPort(), jsonResources);%0A%0A
     

Update KafkaService to start API Server before SchedulerDriver run (#550)