package org.apache.helix.integration;

/*
 * 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.
 */

import java.util.Date;

import org.apache.helix.TestHelper;
import org.apache.helix.ZNRecord;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.mock.controller.ClusterController;
import org.apache.helix.mock.participant.MockParticipant;
import org.apache.helix.mock.participant.MockParticipant.MockMSModelFactory;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
import org.apache.helix.tools.ClusterSetup;
import org.apache.helix.tools.ClusterStateVerifier;
import org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
import org.testng.Assert;
import org.testng.annotations.Test;


public class TestAddStateModelFactoryAfterConnect extends ZkIntegrationTestBase
{
  @Test
  public void testAddStateModelFactoryAfterConnect() throws Exception
  {
    // Logger.getRootLogger().setLevel(Level.INFO);
    String className = TestHelper.getTestClassName();
    String methodName = TestHelper.getTestMethodName();
    String clusterName = className + "_" + methodName;
    final int n = 5;

    System.out.println("START " + clusterName + " at "
        + new Date(System.currentTimeMillis()));

    MockParticipant[] participants = new MockParticipant[n];

    TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
                            "localhost", // participant name prefix
                            "TestDB", // resource name prefix
                            1, // resources
                            10, // partitions per resource
                            n, // number of nodes
                            3, // replicas
                            "MasterSlave",
                            true); // do rebalance

    ClusterController controller =
        new ClusterController(clusterName, "controller_0", ZK_ADDR);
    controller.syncStart();

    // start participants
    for (int i = 0; i < n; i++)
    {
      String instanceName = "localhost_" + (12918 + i);

      participants[i] = new MockParticipant(clusterName, instanceName, ZK_ADDR, null);
      participants[i].syncStart();
    }

    boolean result =
        ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
                                                                                 clusterName));
    Assert.assertTrue(result);

    // add a new idealState without registering message handling factory
    ClusterSetup setupTool = new ClusterSetup(ZK_ADDR);
    setupTool.addResourceToCluster(clusterName, "TestDB1", 16, "MasterSlave");

    ZkBaseDataAccessor<ZNRecord> baseAccessor =
        new ZkBaseDataAccessor<ZNRecord>(_gZkClient);
    ZKHelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);
    Builder keyBuilder = accessor.keyBuilder();
    IdealState idealState = accessor.getProperty(keyBuilder.idealStates("TestDB1"));
    idealState.setStateModelFactoryName("TestDB1_Factory");
    accessor.setProperty(keyBuilder.idealStates("TestDB1"), idealState);
    setupTool.rebalanceStorageCluster(clusterName, "TestDB1", 3);

    // external view for TestDB1 should be empty
    ExternalView extView = null;
    long start = System.currentTimeMillis();
    while (extView == null)
    {
      Thread.sleep(50);
      extView = accessor.getProperty(keyBuilder.externalView("TestDB1"));

      long now = System.currentTimeMillis();
      if (now - start > 5000)
      {
        Assert.fail("Timeout waiting for an empty external view of TestDB1. extView: " + extView);
      }
    }
    Assert.assertEquals(extView.getRecord().getMapFields().size(),
                        0,
                        "External view for TestDB1 should be empty since TestDB1 is added without a state model factory");

    // register "TestDB1_Factory" state model factory
    // Logger.getRootLogger().setLevel(Level.INFO);
    for (int i = 0; i < n; i++)
    {
      participants[i].getManager()
                     .getStateMachineEngine()
                     .registerStateModelFactory("MasterSlave",
                                                new MockMSModelFactory(),
                                                "TestDB1_Factory");
    }

    result =
        ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
                                                                                 clusterName));
    Assert.assertTrue(result);

    // clean up
    // wait for all zk callbacks done
    Thread.sleep(1000);
    controller.syncStop();
    for (int i = 0; i < 5; i++)
    {
      participants[i].syncStop();
    }

    System.out.println("END " + clusterName + " at "
        + new Date(System.currentTimeMillis()));

  }
}

@@ -858,20 +858,71 @@
 va.util.
-Date
+ArrayList;%0Aimport java.util.Date;%0Aimport java.util.List
 ;%0A%0Aimpor
@@ -1336,16 +1336,60 @@
 actory;%0A
+import org.apache.helix.model.CurrentState;%0A
 import o
@@ -1819,40 +1819,13 @@
 test
-AddStateModelFactoryAfterConnect
+Basic
 () t
@@ -4191,75 +4191,35 @@
     
-long start = System.currentTimeMillis();%0A    while (extView == null
+for (int i = 0; i %3C 10; i++
 )%0A  
@@ -4241,17 +4241,18 @@
 d.sleep(
-5
+10
 0);%0A    
@@ -4332,93 +4332,438 @@
     
-long now = System.currentTimeMillis();%0A      if (now - start %3E 5000)%0A      %7B
+if (extView != null)%0A    %09  break;%0A    %7D%0A%0A    if (extView == null) %7B%0A    %09// output current-state for debug%0A    %09List%3CCurrentState%3E curStates = new ArrayList%3CCurrentState%3E();%0A    %09for (int i = 0; i %3C n; i++) %7B%0A    %09%09CurrentState curState = accessor.getProperty(keyBuilder.currentState(participants%5Bi%5D.getInstanceName(), %0A    %09%09%09%09participants%5Bi%5D.getManager().getSessionId(), %22TestDB1%22));%0A    %09%09curStates.add(curState);
 %0A    
+%09%7D%0A
     
+%09
 Asse
@@ -4830,44 +4830,41 @@
 B1. 
-extView: %22 + extView);%0A      %7D%0A    %7D
+curStates: %22 + curStates);%0A    %7D%0A
 %0A   

fix intermittent test failures