/**
* 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.handler;
import com.alibaba.wasp.EntityGroupInfo;
import com.alibaba.wasp.Server;
import com.alibaba.wasp.ServerName;
import com.alibaba.wasp.TableLockedException;
import com.alibaba.wasp.TableNotDisabledException;
import com.alibaba.wasp.TableNotFoundException;
import com.alibaba.wasp.executor.EventHandler;
import com.alibaba.wasp.master.AssignmentManager;
import com.alibaba.wasp.master.BulkAssigner;
import com.alibaba.wasp.master.EntityGroupPlan;
import com.alibaba.wasp.master.EntityGroupStates;
import com.alibaba.wasp.master.FMaster;
import com.alibaba.wasp.master.FMasterServices;
import com.alibaba.wasp.master.FServerManager;
import com.alibaba.wasp.master.TableLockManager;
import com.alibaba.wasp.meta.FMetaReader;
import com.alibaba.wasp.meta.FTable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.zookeeper.KeeperException;
import org.cloudera.htrace.Trace;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
/**
* Handler to run enable of a table.
*/
public class EnableTableHandler extends EventHandler {
private static final Log LOG = LogFactory.getLog(EnableTableHandler.class);
private final byte[] tableName;
private final String tableNameStr;
private final AssignmentManager assignmentManager;
private boolean retainAssignment = false;
private boolean isChild = false;
private TableLockManager tableLockManager;
public EnableTableHandler(Server server, FMasterServices fMasterServices,
AssignmentManager assignmentManager, byte[] tableName,
boolean skipTableStateCheck) throws TableNotFoundException,
TableNotDisabledException, IOException {
super(server, EventType.C_M_ENABLE_TABLE);
this.tableName = tableName;
this.tableNameStr = Bytes.toString(tableName);
this.assignmentManager = assignmentManager;
this.retainAssignment = skipTableStateCheck;
// Check if table exists
if (!FMetaReader.tableExists(server.getConfiguration(), this.tableNameStr)) {
throw new TableNotFoundException(tableNameStr);
}
FTable ftable = FMetaReader.getTable(server.getConfiguration(),
tableNameStr);
// Child Table
if (ftable.isChildTable()) {
isChild = true;
}
// There could be multiple client requests trying to disable or enable
// the table at the same time. Ensure only the first request is honored
// After that, no other requests can be accepted until the table reaches
// DISABLED or ENABLED.
try {
this.tableLockManager = fMasterServices.getTableLockManager();
if (tableLockManager.lockTable(tableNameStr)) {
LOG.info("lock table '" + tableNameStr + "' by EnableTableHandler");
} else {
throw new TableLockedException(tableNameStr + " has been locked. ");
}
if (!skipTableStateCheck) {
try {
if (!this.assignmentManager.getZKTable()
.checkDisabledAndSetEnablingTable(this.tableNameStr)) {
LOG.info("Table " + tableNameStr
+ " isn't disabled; skipping enable");
throw new TableNotDisabledException(this.tableNameStr);
}
} catch (KeeperException e) {
tableLockManager.unlockTable(tableNameStr);
throw new IOException("Unable to ensure that the table will be"
+ " enabling because of a ZooKeeper issue", e);
}
}
} catch (TableNotDisabledException e) {
tableLockManager.unlockTable(tableNameStr);
throw e;
}
}
@Override
public void process() throws IOException {
try {
LOG.info("Attempting to enable the table " + this.tableNameStr);
handleEnableTable();
} catch (IOException e) {
LOG.error("Error trying to enable the table " + this.tableNameStr, e);
} catch (KeeperException e) {
LOG.error("Error trying to enable the table " + this.tableNameStr, e);
} catch (InterruptedException e) {
LOG.error("Error trying to enable the table " + this.tableNameStr, e);
} finally {
tableLockManager.unlockTable(tableNameStr);
}
}
private void handleEnableTable() throws IOException, KeeperException,
InterruptedException {
// I could check table is disabling and if so, not enable but require
// that user first finish disabling but that might be obnoxious.
// Set table enabling flag up in zk.
this.assignmentManager.getZKTable().setEnablingTable(this.tableNameStr);
boolean done = false;
if (isChild) {
done = true;
} else {
// Get the entityGroups of this table. We're done when all listed
// tables are onlined.
List<Pair<EntityGroupInfo, ServerName>> tableEntityGroupsAndLocations = FMetaReader
.getTableEntityGroupsAndLocations(server.getConfiguration(),
tableName, true);
int countOfEntityGroupsInTable = tableEntityGroupsAndLocations.size();
List<EntityGroupInfo> entityGroups = entityGroupsToAssignWithServerName(tableEntityGroupsAndLocations);
int entityGroupsCount = entityGroups.size();
if (entityGroupsCount == 0) {
done = true;
}
LOG.info("Table '" + this.tableNameStr + "' has "
+ countOfEntityGroupsInTable + " entityGroups, of which "
+ entityGroupsCount + " are offline.");
BulkEnabler bd = new BulkEnabler(this.server, entityGroups,
countOfEntityGroupsInTable, this.retainAssignment);
try {
if (bd.bulkAssign()) {
done = true;
}
} catch (InterruptedException e) {
LOG.warn("Enable operation was interrupted when enabling table '"
+ this.tableNameStr + "'");
// Preserve the interrupt.
Thread.currentThread().interrupt();
}
}
// notify server to refresh the table's meta cache.
FServerManager serverManager = ((FMasterServices) server)
.getFServerManager();
for (ServerName serverName : serverManager.getOnlineServersList()) {
serverManager.sendEnableTable(serverName, Bytes.toString(tableName));
}
if (done) {
// Flip the table to enabled.
this.assignmentManager.getZKTable().setEnabledTable(this.tableNameStr);
LOG.info("Table '" + this.tableNameStr
+ "' was successfully enabled. Status: done=" + done);
} else {
LOG.warn("Table '" + this.tableNameStr
+ "' wasn't successfully enabled. Status: done=" + done);
}
}
@Override
public String toString() {
String name = "UnknownServerName";
if (server != null && server.getServerName() != null) {
name = server.getServerName().toString();
}
return getClass().getSimpleName() + "-" + name + "-" + getSeqid() + "-"
+ tableNameStr;
}
/**
* @param entityGroupsInMeta
* @return List of entityGroups neither in transition nor assigned.
* @throws java.io.IOException
*/
private List<EntityGroupInfo> entityGroupsToAssignWithServerName(
final List<Pair<EntityGroupInfo, ServerName>> entityGroupsInMeta)
throws IOException {
FServerManager serverManager = ((FMaster) this.server).getFServerManager();
List<EntityGroupInfo> entityGroups = new ArrayList<EntityGroupInfo>();
EntityGroupStates entityGroupStates = this.assignmentManager
.getEntityGroupStates();
for (Pair<EntityGroupInfo, ServerName> entityGroupLocation : entityGroupsInMeta) {
EntityGroupInfo egi = entityGroupLocation.getFirst();
ServerName sn = entityGroupLocation.getSecond();
if (!entityGroupStates.isEntityGroupInTransition(egi)
&& !entityGroupStates.isEntityGroupAssigned(egi)) {
if (this.retainAssignment && sn != null
&& serverManager.isServerOnline(sn)) {
this.assignmentManager.addPlan(egi.getEncodedName(),
new EntityGroupPlan(egi, null, sn));
}
entityGroups.add(egi);
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("Skipping assign for the entityGroup " + egi
+ " during enable table " + egi.getTableNameAsString()
+ " because its already in tranition or assigned.");
}
}
}
return entityGroups;
}
/**
* Run bulk enable.
*/
class BulkEnabler extends BulkAssigner {
private final List<EntityGroupInfo> entityGroups;
// Count of entityGroups in table at time this assign was launched.
private final int countOfEntityGroupsInTable;
private final boolean retainAssignment;
BulkEnabler(final Server server, final List<EntityGroupInfo> entityGroups,
final int countOfEntityGroupsInTable, boolean retainAssignment) {
super(server);
this.entityGroups = entityGroups;
this.countOfEntityGroupsInTable = countOfEntityGroupsInTable;
this.retainAssignment = retainAssignment;
}
@Override
protected void populatePool(ExecutorService pool) throws IOException {
boolean roundRobinAssignment = this.server.getConfiguration().getBoolean(
"wasp.master.enabletable.roundrobin", false);
// In case of masterRestart always go with single assign. Going thro
// roundRobinAssignment will use bulkassign which may lead to double
// assignment.
if (retainAssignment || !roundRobinAssignment) {
for (EntityGroupInfo entityGroup : entityGroups) {
if (assignmentManager.getEntityGroupStates()
.isEntityGroupInTransition(entityGroup)) {
continue;
}
final EntityGroupInfo egi = entityGroup;
pool.execute(Trace.wrap(new Runnable() {
public void run() {
assignmentManager.assign(egi, true);
}
}));
}
} else {
try {
assignmentManager.assign(entityGroups);
} catch (InterruptedException e) {
LOG.warn("Assignment was interrupted");
Thread.currentThread().interrupt();
}
}
}
@Override
protected boolean waitUntilDone(long timeout) throws InterruptedException {
long startTime = System.currentTimeMillis();
long remaining = timeout;
List<EntityGroupInfo> entityGroups = null;
int lastNumberOfEntityGroups = 0;
while (!server.isStopped() && remaining > 0) {
Thread.sleep(waitingTimeForEvents);
entityGroups = assignmentManager.getEntityGroupStates()
.getEntityGroupsOfTable(tableName);
if (isDone(entityGroups))
break;
// Punt on the timeout as long we make progress
if (entityGroups.size() > lastNumberOfEntityGroups) {
lastNumberOfEntityGroups = entityGroups.size();
timeout += waitingTimeForEvents;
}
remaining = timeout - (System.currentTimeMillis() - startTime);
}
return isDone(entityGroups);
}
private boolean isDone(final List<EntityGroupInfo> entityGroups) {
return entityGroups != null
&& entityGroups.size() >= this.countOfEntityGroupsInTable;
}
}
}