helix-core/src/test/java/org/apache/helix/manager/zk/TestWtCacheAsyncOpMultiThread.java

package org.apache.helix.manager.zk;

/*
 * 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.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;

import org.apache.helix.AccessOption;
import org.apache.helix.PropertyPathBuilder;
import org.apache.helix.TestHelper;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.ZNRecordUpdater;
import org.apache.helix.ZkUnitTestBase;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.testng.Assert;
import org.testng.annotations.Test;

public class TestWtCacheAsyncOpMultiThread extends ZkUnitTestBase {
  class TestCreateZkCacheBaseDataAccessor implements Callable<Boolean> {
    final ZkCacheBaseDataAccessor<ZNRecord> _accessor;
    final String _clusterName;
    final int _id;

    TestCreateZkCacheBaseDataAccessor(ZkCacheBaseDataAccessor<ZNRecord> accessor,
        String clusterName, int id) {
      _accessor = accessor;
      _clusterName = clusterName;
      _id = id;
    }

    @Override
    public Boolean call() throws Exception {
      // create 10 current states in 2 steps
      List<String> paths = new ArrayList<>();
      List<ZNRecord> records = new ArrayList<>();
      for (int j = 0; j < 2; j++) {
        paths.clear();
        records.clear();

        if (_id == 1 && j == 0) {
          // let thread_0 create 0-4
          Thread.sleep(30);
        }

        if (_id == 0 && j == 1) {
          // let thread_1 create 5-9
          Thread.sleep(100);
        }

        for (int i = 0; i < 5; i++) {
          int k = j * 5 + i;
          String path = PropertyPathBuilder.instanceCurrentState(_clusterName, "localhost_8901",
              "session_0", "TestDB" + k);
          ZNRecord record = new ZNRecord("TestDB" + k);

          paths.add(path);
          records.add(record);
        }

        boolean[] success = _accessor.createChildren(paths, records, AccessOption.PERSISTENT);
        // System.out.println("thread-" + _id + " creates " + j + ": " + Arrays.toString(success));

        // create all all sync'ed, so we shall see either all true or all false
        for (int i = 1; i < 5; i++) {
          Assert.assertEquals(success[i], success[0], "Should be either all succeed of all fail");
        }
      }

      return true;
    }
  }

  class TestUpdateZkCacheBaseDataAccessor implements Callable<Boolean> {
    final ZkCacheBaseDataAccessor<ZNRecord> _accessor;
    final String _clusterName;
    final int _id;

    TestUpdateZkCacheBaseDataAccessor(ZkCacheBaseDataAccessor<ZNRecord> accessor,
        String clusterName, int id) {
      _accessor = accessor;
      _clusterName = clusterName;
      _id = id;
    }

    @Override
    public Boolean call() {
      // create 10 current states in 2 steps
      List<String> paths = new ArrayList<>();
      List<DataUpdater<ZNRecord>> updaters = new ArrayList<>();
      for (int j = 0; j < 10; j++) {
        paths.clear();
        updaters.clear();

        for (int i = 0; i < 10; i++) {
          String path = PropertyPathBuilder.instanceCurrentState(_clusterName, "localhost_8901",
              "session_0", "TestDB" + i);

          ZNRecord newRecord = new ZNRecord("TestDB" + i);
          newRecord.setSimpleField("" + j, "" + j);
          DataUpdater<ZNRecord> updater = new ZNRecordUpdater(newRecord);
          paths.add(path);
          updaters.add(updater);
        }

        boolean[] success = _accessor.updateChildren(paths, updaters, AccessOption.PERSISTENT);
        // System.out.println("thread-" + _id + " updates " + j + ": " + Arrays.toString(success));

        for (int i = 0; i < 10; i++) {
          Assert.assertTrue(success[i], "Should be all succeed");
        }
      }

      return true;
    }
  }

  class TestSetZkCacheBaseDataAccessor implements Callable<Boolean> {
    final ZkCacheBaseDataAccessor<ZNRecord> _accessor;
    final String _clusterName;
    final int _id;

    TestSetZkCacheBaseDataAccessor(ZkCacheBaseDataAccessor<ZNRecord> accessor, String clusterName,
        int id) {
      _accessor = accessor;
      _clusterName = clusterName;
      _id = id;
    }

    @Override
    public Boolean call() throws Exception {
      // create 10 current states in 2 steps
      List<String> paths = new ArrayList<>();
      List<ZNRecord> records = new ArrayList<>();
      for (int j = 0; j < 2; j++) {
        paths.clear();
        records.clear();

        if (_id == 1 && j == 0) {
          // let thread_0 create 0-4
          Thread.sleep(30);
        }

        if (_id == 0 && j == 1) {
          // let thread_1 create 5-9
          Thread.sleep(100);
        }

        for (int i = 0; i < 5; i++) {
          int k = j * 5 + i;
          String path = PropertyPathBuilder.externalView(_clusterName, "TestDB" + k);
          ZNRecord record = new ZNRecord("TestDB" + k);

          paths.add(path);
          records.add(record);
        }
        boolean[] success = _accessor.setChildren(paths, records, AccessOption.PERSISTENT);

        for (int i = 0; i < 5; i++) {
          Assert.assertTrue(success[i]);
        }
      }

      return true;
    }
  }

  @Test
  public void testHappyPathZkCacheBaseDataAccessor() {
    String className = TestHelper.getTestClassName();
    String methodName = TestHelper.getTestMethodName();
    String clusterName = className + "_" + methodName;
    System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));

    // init zkCacheDataAccessor
    String curStatePath = PropertyPathBuilder.instanceCurrentState(clusterName, "localhost_8901");
    String extViewPath = PropertyPathBuilder.externalView(clusterName);

    ZkBaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);

    baseAccessor.create(curStatePath, null, AccessOption.PERSISTENT);

    List<String> cachePaths = Arrays.asList(curStatePath, extViewPath);
    ZkCacheBaseDataAccessor<ZNRecord> accessor =
        new ZkCacheBaseDataAccessor<>(baseAccessor, null, cachePaths, null);

    // TestHelper.printCache(accessor._wtCache);
    boolean ret = TestHelper.verifyZkCache(cachePaths, accessor._wtCache._cache, _gZkClient, false);
    Assert.assertTrue(ret, "wtCache doesn't match data on Zk");

    // create 10 current states using 2 threads
    List<Callable<Boolean>> threads = new ArrayList<>();
    for (int i = 0; i < 2; i++) {
      threads.add(new TestCreateZkCacheBaseDataAccessor(accessor, clusterName, i));
    }
    TestHelper.startThreadsConcurrently(threads, 1000);

    // verify wtCache
    // TestHelper.printCache(accessor._wtCache);
    ret = TestHelper.verifyZkCache(cachePaths, accessor._wtCache._cache, _gZkClient, false);
    Assert.assertTrue(ret, "wtCache doesn't match data on Zk");

    // update 10 current states 10 times using 2 threads
    threads.clear();
    for (int i = 0; i < 2; i++) {
      threads.add(new TestUpdateZkCacheBaseDataAccessor(accessor, clusterName, i));
    }
    TestHelper.startThreadsConcurrently(threads, 1000);

    // verify wtCache
    // TestHelper.printCache(accessor._wtCache);
    ret = TestHelper.verifyZkCache(cachePaths, accessor._wtCache._cache, _gZkClient, false);
    Assert.assertTrue(ret, "wtCache doesn't match data on Zk");

    // set 10 external views using 2 threads
    threads.clear();
    for (int i = 0; i < 2; i++) {
      threads.add(new TestSetZkCacheBaseDataAccessor(accessor, clusterName, i));
    }
    TestHelper.startThreadsConcurrently(threads, 1000);

    // verify wtCache
    // TestHelper.printCache(accessor._wtCache);
    ret = TestHelper.verifyZkCache(cachePaths, accessor._wtCache._cache, _gZkClient, false);
    Assert.assertTrue(ret, "wtCache doesn't match data on Zk");

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