package org.apache.solr.crossdc;

import com.carrotsearch.randomizedtesting.annotations.ThreadLeakAction;
import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
import com.carrotsearch.randomizedtesting.annotations.ThreadLeakLingering;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.lucene.util.QuickPatchThreadsFilter;
import org.apache.solr.SolrIgnoredThreadsFilter;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.BaseCloudSolrClient;
import org.apache.solr.client.solrj.impl.BaseHttpSolrClient;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.cloud.MiniSolrCloudCluster;
import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.util.ObjectReleaseTracker;
import org.apache.solr.crossdc.common.KafkaCrossDcConf;
import org.apache.solr.crossdc.common.MirroredSolrRequest;
import org.apache.solr.crossdc.common.MirroredSolrRequestSerializer;
import org.apache.solr.crossdc.consumer.Consumer;
import org.junit.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.sys.Prop;

import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import static org.apache.solr.crossdc.common.KafkaCrossDcConf.DEFAULT_MAX_REQUEST_SIZE;
import static org.apache.solr.crossdc.common.KafkaCrossDcConf.INDEX_UNMIRRORABLE_DOCS;
import static org.mockito.Mockito.spy;

@ThreadLeakFilters(defaultFilters = true, filters = { SolrIgnoredThreadsFilter.class,
    QuickPatchThreadsFilter.class, SolrKafkaTestsIgnoredThreadsFilter.class })
@ThreadLeakLingering(linger = 5000) public class SolrAndKafkaIntegrationTest extends
    SolrTestCaseJ4 {

  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

  private static final int MAX_DOC_SIZE_BYTES = Integer.parseInt(DEFAULT_MAX_REQUEST_SIZE);

  static final String VERSION_FIELD = "_version_";

  private static final int NUM_BROKERS = 1;
  public EmbeddedKafkaCluster kafkaCluster;

  protected volatile MiniSolrCloudCluster solrCluster1;
  protected volatile MiniSolrCloudCluster solrCluster2;

  protected static volatile Consumer consumer;

  private static String TOPIC = "topic1";

  private static String COLLECTION = "collection1";
  private static String ALT_COLLECTION = "collection2";

  @Before
  public void beforeSolrAndKafkaIntegrationTest() throws Exception {
    consumer = new Consumer();
    Properties config = new Properties();
    //config.put("unclean.leader.election.enable", "true");
    //config.put("enable.partition.eof", "false");

    kafkaCluster = new EmbeddedKafkaCluster(NUM_BROKERS, config) {
      public String bootstrapServers() {
        return super.bootstrapServers().replaceAll("localhost", "127.0.0.1");
      }
    };
    kafkaCluster.start();

    kafkaCluster.createTopic(TOPIC, 1, 1);

    System.setProperty("topicName", TOPIC);
    System.setProperty("bootstrapServers", kafkaCluster.bootstrapServers());
    System.setProperty(INDEX_UNMIRRORABLE_DOCS, "false");

    solrCluster1 = new SolrCloudTestCase.Builder(1, createTempDir()).addConfig("conf",
        getFile("src/test/resources/configs/cloud-minimal/conf").toPath()).configure();

    CollectionAdminRequest.Create create =
        CollectionAdminRequest.createCollection(COLLECTION, "conf", 1, 1);
    solrCluster1.getSolrClient().request(create);
    solrCluster1.waitForActiveCollection(COLLECTION, 1, 1);

    solrCluster1.getSolrClient().setDefaultCollection(COLLECTION);

    solrCluster2 = new SolrCloudTestCase.Builder(1, createTempDir()).addConfig("conf",
        getFile("src/test/resources/configs/cloud-minimal/conf").toPath()).configure();

    CollectionAdminRequest.Create create2 =
        CollectionAdminRequest.createCollection(COLLECTION, "conf", 1, 1);
    solrCluster2.getSolrClient().request(create2);
    solrCluster2.waitForActiveCollection(COLLECTION, 1, 1);

    solrCluster2.getSolrClient().setDefaultCollection(COLLECTION);

    String bootstrapServers = kafkaCluster.bootstrapServers();
    log.info("bootstrapServers={}", bootstrapServers);

    Map<String, Object> properties = new HashMap<>();
    properties.put(KafkaCrossDcConf.BOOTSTRAP_SERVERS, bootstrapServers);
    properties.put(KafkaCrossDcConf.ZK_CONNECT_STRING, solrCluster2.getZkServer().getZkAddress());
    properties.put(KafkaCrossDcConf.TOPIC_NAME, TOPIC);
    properties.put(KafkaCrossDcConf.GROUP_ID, "group1");
    properties.put(KafkaCrossDcConf.MAX_REQUEST_SIZE_BYTES, MAX_DOC_SIZE_BYTES);
    consumer.start(properties);

  }

  @After
  public void afterSolrAndKafkaIntegrationTest() throws Exception {
    ObjectReleaseTracker.clear();

    if (solrCluster1 != null) {
      solrCluster1.getZkServer().getZkClient().printLayoutToStdOut();
      solrCluster1.shutdown();
    }
    if (solrCluster2 != null) {
      solrCluster2.getZkServer().getZkClient().printLayoutToStdOut();
      solrCluster2.shutdown();
    }

    consumer.shutdown();
    consumer = null;

    try {
      //kafkaCluster.deleteAllTopicsAndWait(10000);
      kafkaCluster.stop();
      kafkaCluster = null;
    } catch (Exception e) {
      log.error("Exception stopping Kafka cluster", e);
    }


  }

//  public void testFullCloudToCloud() throws Exception {
//    CloudSolrClient client = solrCluster1.getSolrClient();
//    SolrInputDocument doc = new SolrInputDocument();
//    doc.addField("id", String.valueOf(System.currentTimeMillis()));
//    doc.addField("text", "some test");
//
//    client.add(doc);
//
//    client.commit(COLLECTION);
//
//    System.out.println("Sent producer record");
//
//    assertCluster2EventuallyHasDocs(COLLECTION, "*:*", 1);
//  }
//
  private static SolrInputDocument getDoc() {
    SolrInputDocument doc = new SolrInputDocument();
    doc.addField("id", String.valueOf(System.nanoTime()));
    doc.addField("text", "some test");
    return doc;
  }
//
//  public void testProducerToCloud() throws Exception {
//    Properties properties = new Properties();
//    properties.put("bootstrap.servers", kafkaCluster.bootstrapServers());
//    properties.put("acks", "all");
//    properties.put("retries", 1);
//    properties.put("batch.size", 1);
//    properties.put("buffer.memory", 33554432);
//    properties.put("linger.ms", 1);
//    properties.put("key.serializer", StringSerializer.class.getName());
//    properties.put("value.serializer", MirroredSolrRequestSerializer.class.getName());
//    Producer<String, MirroredSolrRequest> producer = new KafkaProducer(properties);
//    UpdateRequest updateRequest = new UpdateRequest();
//    updateRequest.setParam("shouldMirror", "true");
//    updateRequest.add("id", String.valueOf(System.currentTimeMillis()), "text", "test");
//    updateRequest.add("id", String.valueOf(System.currentTimeMillis() + 22), "text", "test2");
//    updateRequest.setParam("collection", COLLECTION);
//    MirroredSolrRequest mirroredSolrRequest = new MirroredSolrRequest(updateRequest);
//    System.out.println("About to send producer record");
//    producer.send(new ProducerRecord(TOPIC, mirroredSolrRequest), (metadata, exception) -> {
//      log.info("Producer finished sending metadata={}, exception={}", metadata, exception);
//    });
//    producer.flush();
//
//    System.out.println("Sent producer record");
//
//    solrCluster2.getSolrClient().commit(COLLECTION);
//
//    assertCluster2EventuallyHasDocs(COLLECTION, "*:*", 2);
//
//    producer.close();
//  }

//  @Test
//  public void testMirroringUpdateProcessor() throws Exception {
//    final SolrInputDocument tooLargeDoc = new SolrInputDocument();
//    tooLargeDoc.addField("id", "tooLarge-" + String.valueOf(System.currentTimeMillis()));
//    tooLargeDoc.addField("text", new String(new byte[2 * MAX_DOC_SIZE_BYTES]));
//    final SolrInputDocument normalDoc = new SolrInputDocument();
//    normalDoc.addField("id", "normalDoc-" + String.valueOf(System.currentTimeMillis()));
//    normalDoc.addField("text", "Hello world");
//    final List<SolrInputDocument> docsToIndex = new ArrayList<>();
//    docsToIndex.add(normalDoc);
//    docsToIndex.add(tooLargeDoc);
//
//    final CloudSolrClient cluster1Client = solrCluster1.getSolrClient();
//    try {
//      cluster1Client.add(docsToIndex);
//    } catch (BaseCloudSolrClient.RouteException e) {
//      // expected
//    }
//    cluster1Client.commit(COLLECTION);
//
//    // Primary and secondary should each only index 'normalDoc'
//    final String normalDocQuery = "id:" + normalDoc.get("id").getFirstValue();
//    assertCluster2EventuallyHasDocs(COLLECTION, normalDocQuery, 1);
//    assertCluster2EventuallyHasDocs(COLLECTION, "*:*", 1);
//    assertClusterEventuallyHasDocs(cluster1Client, COLLECTION, normalDocQuery, 1);
//    assertClusterEventuallyHasDocs(cluster1Client, COLLECTION, "*:*", 1);
//
//    // Create new primary+secondary collection where 'tooLarge' docs ARE indexed on the primary
//    CollectionAdminRequest.Create create =
//        CollectionAdminRequest.createCollection(ALT_COLLECTION, "conf", 1, 1)
//            .withProperty("indexUnmirrorableDocs", "true");
//    try {
//      solrCluster1.getSolrClient().request(create);
//      solrCluster2.getSolrClient().request(create);
//      solrCluster1.waitForActiveCollection(ALT_COLLECTION, 1, 1);
//      solrCluster2.waitForActiveCollection(ALT_COLLECTION, 1, 1);
//
//      cluster1Client.add(ALT_COLLECTION, docsToIndex);
//      cluster1Client.commit(ALT_COLLECTION);
//
//      // Primary should have both 'normal' and 'large' docs; secondary should only have 'normal' doc.
//      assertClusterEventuallyHasDocs(cluster1Client, ALT_COLLECTION, "*:*", 2);
//      assertCluster2EventuallyHasDocs(ALT_COLLECTION, normalDocQuery, 1);
//      assertCluster2EventuallyHasDocs(ALT_COLLECTION, "*:*", 1);
//    } finally {
//      CollectionAdminRequest.Delete delete =
//        CollectionAdminRequest.deleteCollection(ALT_COLLECTION);
//      solrCluster1.getSolrClient().request(delete);
//      solrCluster2.getSolrClient().request(delete);
//    }
//  }

  private void assertCluster2EventuallyHasDocs(String collection, String query, int expectedNumDocs) throws Exception {
    assertClusterEventuallyHasDocs(solrCluster2.getSolrClient(), collection, query, expectedNumDocs);
  }

  private void createCollection(CloudSolrClient client, CollectionAdminRequest.Create createCmd) throws Exception {
    final String stashedDefault = client.getDefaultCollection();
    try {
      //client.setDefaultCollection(null);
      client.request(createCmd);
    } finally {
      //client.setDefaultCollection(stashedDefault);
    }
  }

  @Test
  public void testFullCloudToCloudMultiCollection() throws Exception {
    CollectionAdminRequest.Create create =
            CollectionAdminRequest.createCollection(ALT_COLLECTION, "conf", 1, 1);

    try {
      solrCluster1.getSolrClient().request(create);
      solrCluster1.waitForActiveCollection(ALT_COLLECTION, 1, 1);

      solrCluster2.getSolrClient().request(create);
      solrCluster2.waitForActiveCollection(ALT_COLLECTION, 1, 1);


      CloudSolrClient client = solrCluster1.getSolrClient();

      SolrInputDocument doc1 = getDoc();
      SolrInputDocument doc2 = getDoc();
      SolrInputDocument doc3 = getDoc();
      SolrInputDocument doc4 = getDoc();
      SolrInputDocument doc5 = getDoc();
      SolrInputDocument doc6 = getDoc();
      SolrInputDocument doc7 = getDoc();

      client.add(COLLECTION, doc1);
      client.add(ALT_COLLECTION, doc2);
      client.add(COLLECTION, doc3);
      client.add(COLLECTION, doc4);
      client.add(ALT_COLLECTION, doc5);
      client.add(ALT_COLLECTION, doc6);
      client.add(COLLECTION, doc7);

      client.commit(COLLECTION);
      client.commit(ALT_COLLECTION);

      System.out.println("Sent producer record");

      assertCluster2EventuallyHasDocs(ALT_COLLECTION, "*:*", 3);
      assertCluster2EventuallyHasDocs(COLLECTION, "*:*", 4);

    } finally {
      CollectionAdminRequest.Delete delete =
              CollectionAdminRequest.deleteCollection(ALT_COLLECTION);
      solrCluster1.getSolrClient().request(delete);
      solrCluster2.getSolrClient().request(delete);
    }
  }


  private void assertClusterEventuallyHasDocs(SolrClient client, String collection, String query, int expectedNumDocs) throws Exception {
    QueryResponse results = null;
    boolean foundUpdates = false;
    for (int i = 0; i < 100; i++) {
      client.commit(collection);
      results = client.query(collection, new SolrQuery(query));
      if (results.getResults().getNumFound() == expectedNumDocs) {
        foundUpdates = true;
      } else {
        Thread.sleep(200);
      }
    }

    assertTrue("results=" + results, foundUpdates);
  }
}