/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.solr.cloud;

import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.solr.JSONTestUtil;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.cloud.SocketProxy;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.QueryRequest;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.TimeSource;
import org.apache.solr.util.TimeOut;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TestCloudConsistency extends SolrCloudTestCase {

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

  private static Map<JettySolrRunner, SocketProxy> proxies;
  private static Map<URI, JettySolrRunner> jettys;

  @Before
  public void setupCluster() throws Exception {
    System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
    System.setProperty("solr.ulog.numRecordsToKeep", "1000");
    System.setProperty("leaderVoteWait", "60000");

    configureCluster(4)
        .addConfig("conf", configset("cloud-minimal"))
        .configure();
    // Add proxies
    proxies = new HashMap<>(cluster.getJettySolrRunners().size());
    jettys = new HashMap<>();
    for (JettySolrRunner jetty:cluster.getJettySolrRunners()) {
      SocketProxy proxy = new SocketProxy();
      jetty.setProxyPort(proxy.getListenPort());
      cluster.stopJettySolrRunner(jetty);//TODO: Can we avoid this restart
      cluster.startJettySolrRunner(jetty);
      proxy.open(jetty.getBaseUrl().toURI());
      log.info("Adding proxy for URL: " + jetty.getBaseUrl() + ". Proxy: " + proxy.getUrl());
      proxies.put(jetty, proxy);
      jettys.put(proxy.getUrl(), jetty);
    }
  }

  @After
  public void tearDownCluster() throws Exception {
    for (SocketProxy proxy:proxies.values()) {
      proxy.close();
    }
    proxies = null;
    jettys = null;
    System.clearProperty("solr.directoryFactory");
    System.clearProperty("solr.ulog.numRecordsToKeep");
    System.clearProperty("leaderVoteWait");
    
    shutdownCluster();
  }

  @Test
  //commented 2-Aug-2018 @LuceneTestCase.BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 12-Jun-2018
  @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 6-Sep-2018
  public void testOutOfSyncReplicasCannotBecomeLeader() throws Exception {
    testOutOfSyncReplicasCannotBecomeLeader(false);
  }

  @Test
  // commented out on: 24-Dec-2018   @LuceneTestCase.BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 2-Aug-2018
  public void testOutOfSyncReplicasCannotBecomeLeaderAfterRestart() throws Exception {
    testOutOfSyncReplicasCannotBecomeLeader(true);
  }

  public void testOutOfSyncReplicasCannotBecomeLeader(boolean onRestart) throws Exception {
    final String collectionName = "outOfSyncReplicasCannotBecomeLeader-"+onRestart;
    CollectionAdminRequest.createCollection(collectionName, 1, 3)
        .setCreateNodeSet("")
        .process(cluster.getSolrClient());
    CollectionAdminRequest.addReplicaToShard(collectionName, "shard1")
        .setNode(cluster.getJettySolrRunner(0).getNodeName())
        .process(cluster.getSolrClient());
    waitForState("Timeout waiting for shard leader", collectionName, clusterShape(1, 1));

    CollectionAdminRequest.addReplicaToShard(collectionName, "shard1")
        .setNode(cluster.getJettySolrRunner(1).getNodeName())
        .process(cluster.getSolrClient());
    CollectionAdminRequest.addReplicaToShard(collectionName, "shard1")
        .setNode(cluster.getJettySolrRunner(2).getNodeName())
        .process(cluster.getSolrClient());
    
    cluster.waitForActiveCollection(collectionName, 1, 3);
    
    waitForState("Timeout waiting for 1x3 collection", collectionName, clusterShape(1, 3));

    addDocs(collectionName, 3, 1);

    final Replica oldLeader = getCollectionState(collectionName).getSlice("shard1").getLeader();
    assertEquals(cluster.getJettySolrRunner(0).getNodeName(), oldLeader.getNodeName());

    if (onRestart) {
      addDocToWhenOtherReplicasAreDown(collectionName, oldLeader, 4);
    } else {
      addDocWhenOtherReplicasAreNetworkPartitioned(collectionName, oldLeader, 4);
    }

    assertDocsExistInAllReplicas(getCollectionState(collectionName).getReplicas(), collectionName, 1, 4);

    CollectionAdminRequest.deleteCollection(collectionName).process(cluster.getSolrClient());
  }


  /**
   * Adding doc when replicas (not leader) are down,
   * These replicas are out-of-sync hence they should not become leader even when current leader is DOWN.
   * Leader should be on node - 0
   */
  private void addDocToWhenOtherReplicasAreDown(String collection, Replica leader, int docId) throws Exception {
    JettySolrRunner j1 = cluster.getJettySolrRunner(1);
    JettySolrRunner j2 = cluster.getJettySolrRunner(2);
    j1.stop();
    j2.stop();
    cluster.waitForJettyToStop(j1);
    cluster.waitForJettyToStop(j2);
    
    waitForState("", collection, (liveNodes, collectionState) ->
      collectionState.getSlice("shard1").getReplicas().stream()
          .filter(replica -> replica.getState() == Replica.State.DOWN).count() == 2);

    addDocs(collection, 1, docId);
    JettySolrRunner j3 = cluster.getJettySolrRunner(0);
    j3.stop();
    cluster.waitForJettyToStop(j3);
    waitForState("", collection, (liveNodes, collectionState) -> collectionState.getReplica(leader.getName()).getState() == Replica.State.DOWN);

    cluster.getJettySolrRunner(1).start();
    cluster.getJettySolrRunner(2).start();
    
    cluster.waitForNode(j1, 30);
    cluster.waitForNode(j2, 30);
    
    TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME);
    while (!timeOut.hasTimedOut()) {
      Replica newLeader = getCollectionState(collection).getSlice("shard1").getLeader();
      if (newLeader != null && !newLeader.getName().equals(leader.getName()) && newLeader.getState() == Replica.State.ACTIVE) {
        fail("Out of sync replica became leader " + newLeader);
      }
    }

    JettySolrRunner j0 = cluster.getJettySolrRunner(0);
    j0.start();
    cluster.waitForNode(j0, 30);
    
    // waitForNode not solid yet?
    cluster.waitForAllNodes(30);
    
    waitForState("Timeout waiting for leader", collection, (liveNodes, collectionState) -> {
      Replica newLeader = collectionState.getLeader("shard1");
      return newLeader != null && newLeader.getName().equals(leader.getName());
    });
    waitForState("Timeout waiting for active collection", collection, clusterShape(1, 3));
  }


  /**
   * Adding doc when replicas (not leader) are network partitioned with leader,
   * These replicas are out-of-sync hence they should not become leader even when current leader is DOWN.
   * Leader should be on node - 0
   */
  private void addDocWhenOtherReplicasAreNetworkPartitioned(String collection, Replica leader, int docId) throws Exception {
    for (int i = 0; i < 3; i++) {
      proxies.get(cluster.getJettySolrRunner(i)).close();
    }
    addDoc(collection, docId, cluster.getJettySolrRunner(0));
    JettySolrRunner j1 = cluster.getJettySolrRunner(0);
    j1.stop();
    cluster.waitForJettyToStop(j1);
    for (int i = 1; i < 3; i++) {
      proxies.get(cluster.getJettySolrRunner(i)).reopen();
    }
    waitForState("Timeout waiting for leader goes DOWN", collection, (liveNodes, collectionState)
        -> collectionState.getReplica(leader.getName()).getState() == Replica.State.DOWN);

    TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME);
    while (!timeOut.hasTimedOut()) {
      Replica newLeader = getCollectionState(collection).getLeader("shard1");
      if (newLeader != null && !newLeader.getName().equals(leader.getName()) && newLeader.getState() == Replica.State.ACTIVE) {
        fail("Out of sync replica became leader " + newLeader);
      }
    }

    proxies.get(cluster.getJettySolrRunner(0)).reopen();
    cluster.getJettySolrRunner(0).start();
    cluster.waitForAllNodes(30);;
    waitForState("Timeout waiting for leader", collection, (liveNodes, collectionState) -> {
      Replica newLeader = collectionState.getLeader("shard1");
      return newLeader != null && newLeader.getName().equals(leader.getName());
    });
    waitForState("Timeout waiting for active collection", collection, clusterShape(1, 3));
    
    cluster.waitForActiveCollection(collection, 1, 3);
  }

  private void addDocs(String collection, int numDocs, int startId) throws SolrServerException, IOException {
    List<SolrInputDocument> docs = new ArrayList<>(numDocs);
    for (int i = 0; i < numDocs; i++) {
      int id = startId + i;
      docs.add(new SolrInputDocument("id", String.valueOf(id), "fieldName_s", String.valueOf(id)));
    }
    cluster.getSolrClient().add(collection, docs);
    cluster.getSolrClient().commit(collection);
  }

  private void addDoc(String collection, int docId, JettySolrRunner solrRunner) throws IOException, SolrServerException {
    try (HttpSolrClient solrClient = new HttpSolrClient.Builder(solrRunner.getBaseUrl().toString()).build()) {
      solrClient.add(collection, new SolrInputDocument("id", String.valueOf(docId), "fieldName_s", String.valueOf(docId)));
    }
  }

  private void assertDocsExistInAllReplicas(List<Replica> notLeaders,
                                              String testCollectionName, int firstDocId, int lastDocId) throws Exception {
    Replica leader =
        cluster.getSolrClient().getZkStateReader().getLeaderRetry(testCollectionName, "shard1", 10000);
    HttpSolrClient leaderSolr = getHttpSolrClient(leader, testCollectionName);
    List<HttpSolrClient> replicas =
        new ArrayList<HttpSolrClient>(notLeaders.size());

    for (Replica r : notLeaders) {
      replicas.add(getHttpSolrClient(r, testCollectionName));
    }
    try {
      for (int d = firstDocId; d <= lastDocId; d++) {
        String docId = String.valueOf(d);
        assertDocExists(leaderSolr, testCollectionName, docId);
        for (HttpSolrClient replicaSolr : replicas) {
          assertDocExists(replicaSolr, testCollectionName, docId);
        }
      }
    } finally {
      if (leaderSolr != null) {
        leaderSolr.close();
      }
      for (HttpSolrClient replicaSolr : replicas) {
        replicaSolr.close();
      }
    }
  }

  private void assertDocExists(HttpSolrClient solr, String coll, String docId) throws Exception {
    NamedList rsp = realTimeGetDocId(solr, docId);
    String match = JSONTestUtil.matchObj("/id", rsp.get("doc"), docId);
    assertTrue("Doc with id=" + docId + " not found in " + solr.getBaseURL()
        + " due to: " + match + "; rsp="+rsp, match == null);
  }

  private NamedList realTimeGetDocId(HttpSolrClient solr, String docId) throws SolrServerException, IOException {
    QueryRequest qr = new QueryRequest(params("qt", "/get", "id", docId, "distrib", "false"));
    return solr.request(qr);
  }

  protected HttpSolrClient getHttpSolrClient(Replica replica, String coll) throws Exception {
    ZkCoreNodeProps zkProps = new ZkCoreNodeProps(replica);
    String url = zkProps.getBaseUrl() + "/" + coll;
    return getHttpSolrClient(url);
  }

}
