/**
* Copyright The Apache Software Foundation
*
* 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 com.alibaba.wasp.master;
import com.alibaba.wasp.EntityGroupInfo;
import com.alibaba.wasp.FConstants;
import com.alibaba.wasp.ServerName;
import com.alibaba.wasp.WaspTestingUtility;
import com.alibaba.wasp.client.WaspAdmin;
import com.alibaba.wasp.fserver.FServer;
import com.alibaba.wasp.fserver.redo.MemRedoLog;
import com.alibaba.wasp.fserver.redo.Redo;
import com.alibaba.wasp.meta.FMetaEditor;
import com.alibaba.wasp.meta.FMetaTestUtil;
import com.alibaba.wasp.meta.FTable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.io.IOException;
import java.util.List;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class TestAssignmentManagerOnCluster {
private final static WaspTestingUtility TEST_UTIL = new WaspTestingUtility();
private final static Configuration conf = TEST_UTIL.getConfiguration();
private static WaspAdmin admin;
@BeforeClass
public static void setUpBeforeClass() throws Exception {
TEST_UTIL.getConfiguration().setClass(FConstants.REDO_IMPL,
MemRedoLog.class, Redo.class);
TEST_UTIL.startMiniCluster(3);
admin = TEST_UTIL.getWaspAdmin();
}
@AfterClass
public static void tearDownAfterClass() throws Exception {
TEST_UTIL.shutdownMiniCluster();
}
/**
* This tests entityGroup assignment
*/
@Test(timeout = 120000)
public void testAssignEntityGroup() throws Exception {
String table = "testAssignEntityGroup";
try {
FTable desc = FMetaTestUtil.makeTable(table);
admin.createTable(desc);
EntityGroupInfo egInfo = new EntityGroupInfo(Bytes.toBytes(desc
.getTableName()), Bytes.toBytes("A"), Bytes.toBytes("Z"));
FMetaEditor.addEntityGroupToMeta(conf, egInfo);
FMaster master = TEST_UTIL.getWaspCluster().getMaster();
master.assignEntityGroup(egInfo);
master.getAssignmentManager().waitForAssignment(egInfo);
ServerName serverName = master.getAssignmentManager()
.getEntityGroupStates().getFServerOfEntityGroup(egInfo);
TEST_UTIL.assertEntityGroupOnServer(egInfo, serverName, 200);
} finally {
TEST_UTIL.deleteTable(Bytes.toBytes(table));
}
}
/**
* This tests offlining a entityGroup
*/
@Test(timeout = 120000)
public void testOfflineEntityGroup() throws Exception {
String table = "testOfflineEntityGroup";
try {
EntityGroupInfo egInfo = createTableAndGetOneEntityGroup(table);
EntityGroupStates entityGroupStates = TEST_UTIL.getWaspCluster()
.getMaster().getAssignmentManager().getEntityGroupStates();
ServerName serverName = entityGroupStates.getFServerOfEntityGroup(egInfo);
TEST_UTIL.assertEntityGroupOnServer(egInfo, serverName, 200);
admin.offline(egInfo.getEntityGroupName());
long timeoutTime = System.currentTimeMillis() + 800;
while (true) {
List<EntityGroupInfo> entityGroups = entityGroupStates
.getEntityGroupsOfTable(Bytes.toBytes(table));
if (!entityGroups.contains(egInfo))
break;
long now = System.currentTimeMillis();
if (now > timeoutTime) {
fail("Failed to offline the entityGroup in time");
break;
}
Thread.sleep(10);
}
EntityGroupState entityGroupState = entityGroupStates
.getEntityGroupState(egInfo);
assertTrue(entityGroupState.isOffline());
} finally {
TEST_UTIL.deleteTable(Bytes.toBytes(table));
}
}
/**
* This tests moving a entityGroup
*/
@Test(timeout = 120000)
public void testMoveEntityGroup() throws Exception {
String table = "testMoveEntityGroup";
try {
EntityGroupInfo egInfo = createTableAndGetOneEntityGroup(table);
EntityGroupStates entityGroupStates = TEST_UTIL.getWaspCluster()
.getMaster().getAssignmentManager().getEntityGroupStates();
ServerName serverName = entityGroupStates.getFServerOfEntityGroup(egInfo);
ServerName destServerName = null;
for (int i = 0; i < 3; i++) {
FServer destServer = TEST_UTIL.getWaspCluster().getFServer(i);
if (!destServer.getServerName().equals(serverName)) {
destServerName = destServer.getServerName();
break;
}
}
assertTrue(destServerName != null && !destServerName.equals(serverName));
TEST_UTIL.getWaspAdmin().move(egInfo.getEncodedNameAsBytes(),
Bytes.toBytes(destServerName.getServerName()));
long timeoutTime = System.currentTimeMillis() + 5000;
while (true) {
ServerName sn = entityGroupStates.getFServerOfEntityGroup(egInfo);
if (sn != null && sn.equals(destServerName)) {
TEST_UTIL.assertEntityGroupOnServer(egInfo, sn, 2000);
break;
}
long now = System.currentTimeMillis();
if (now > timeoutTime) {
fail("Failed to move the entityGroup in time");
}
entityGroupStates.waitForUpdate(50);
}
} finally {
TEST_UTIL.deleteTable(Bytes.toBytes(table));
}
}
EntityGroupInfo createTableAndGetOneEntityGroup(final String tableName)
throws IOException, InterruptedException {
FTable desc = FMetaTestUtil.makeTable(tableName);
admin.createTable(desc, Bytes.toBytes("A"), Bytes.toBytes("Z"), 5);
// wait till the table is assigned
FMaster master = TEST_UTIL.getWaspCluster().getMaster();
long timeoutTime = System.currentTimeMillis() + 100;
while (true) {
List<EntityGroupInfo> entityGroups = master.getAssignmentManager()
.getEntityGroupStates()
.getEntityGroupsOfTable(Bytes.toBytes(tableName));
if (entityGroups.size() > 3) {
return entityGroups.get(2);
}
long now = System.currentTimeMillis();
if (now > timeoutTime) {
fail("Could not find an online entityGroup");
}
Thread.sleep(10);
}
}
}