View Javadoc

1   package org.apache.helix.integration;
2   
3   /*
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *   http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing,
15   * software distributed under the License is distributed on an
16   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
17   * KIND, either express or implied.  See the License for the
18   * specific language governing permissions and limitations
19   * under the License.
20   */
21  
22  import java.util.List;
23  
24  import org.apache.helix.HelixDataAccessor;
25  import org.apache.helix.PropertyKey.Builder;
26  import org.apache.helix.controller.restlet.ZKPropertyTransferServer;
27  import org.apache.helix.controller.restlet.ZkPropertyTransferClient;
28  import org.apache.helix.model.StatusUpdate;
29  import org.testng.Assert;
30  import org.testng.annotations.AfterClass;
31  import org.testng.annotations.BeforeClass;
32  
33  /**
34   * 
35   * setup a storage cluster and start a zk-based cluster controller in stand-alone mode
36   * start 5 dummy participants verify the current states at end
37   */
38  
39  public class ZkStandAloneCMTestBaseWithPropertyServerCheck extends ZkStandAloneCMTestBase
40  {
41    @BeforeClass
42    public void beforeClass() throws Exception
43    {
44      ZKPropertyTransferServer.PERIOD = 500;
45      ZkPropertyTransferClient.SEND_PERIOD = 500;
46      ZKPropertyTransferServer.getInstance().init(19999, ZK_ADDR);
47      super.beforeClass();
48      
49      Thread.sleep(1000);
50      for (int i = 0; i < NODE_NR; i++)
51      {
52        String instanceName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
53        if (_startCMResultMap.get(instanceName) != null)
54        {
55          HelixDataAccessor accessor = _startCMResultMap.get(instanceName)._manager.getHelixDataAccessor();
56          Builder kb = accessor.keyBuilder();
57          List<StatusUpdate> statusUpdates = accessor.getChildValues(
58              kb.stateTransitionStatus(instanceName, _startCMResultMap.get(instanceName)._manager.getSessionId(),
59                  TEST_DB));
60          for(int j = 0;j < 10; j++)
61          {
62            statusUpdates = accessor.getChildValues(
63              kb.stateTransitionStatus(instanceName, _startCMResultMap.get(instanceName)._manager.getSessionId(),
64                  TEST_DB));
65            if(statusUpdates.size() == 0)
66            {
67              Thread.sleep(500);
68            }
69            else
70            {
71              break;
72            }
73          }
74          Assert.assertTrue(statusUpdates.size() > 0);
75          for(StatusUpdate update : statusUpdates)
76          {
77            Assert.assertTrue(update.getRecord().getSimpleField(ZkPropertyTransferClient.USE_PROPERTYTRANSFER).equals("true"));
78            Assert.assertTrue(update.getRecord().getSimpleField(ZKPropertyTransferServer.SERVER) != null);
79          }
80        }
81      }
82    }
83  
84    @AfterClass
85    public void afterClass() throws Exception
86    {
87      super.afterClass();
88      ZKPropertyTransferServer.getInstance().shutdown();
89      ZKPropertyTransferServer.getInstance().reset();
90    }
91  }