/**
*
* 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.executor;
import com.alibaba.wasp.Server;
import com.alibaba.wasp.executor.ExecutorService.ExecutorType;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.cloudera.htrace.Sampler;
import org.cloudera.htrace.Span;
import org.cloudera.htrace.Trace;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
/**
* Abstract base class for all wasp event handlers. Subclasses should implement
* the {@link #process()} method. Subclasses should also do all necessary checks
* up in their constructor if possible -- check table exists, is disabled, etc.
* -- so they fail fast rather than later when process is running. Do it this
* way because process be invoked directly but event handlers are also run in an
* executor context -- i.e. asynchronously -- and in this case, exceptions
* thrown at process time will not be seen by the invoker, not till we implement
* a call-back mechanism so the client can pick them up later.
* <p>
* Event handlers have an {@link EventType}. {@link EventType} is a list of ALL
* handler event types. We need to keep a full list in one place -- and as enums
* is a good shorthand for an implemenations -- because event handlers can be
* passed to executors when they are to be run asynchronously. The wasp
* executor, see {@link ExecutorService}, has a switch for passing event type to
* executor.
* <p>
* Event listeners can be installed and will be called pre- and post- process if
* this EventHandler is run in a Thread (its a Runnable so if its {@link #run()}
* method gets called). Implement {@link EventHandlerListener}s, and registering
* using {@link #setListener(EventHandlerListener)}.
*
* @see ExecutorService
*/
public abstract class EventHandler implements Runnable, Comparable<Runnable> {
private static final Log LOG = LogFactory.getLog(EventHandler.class);
// type of event this object represents
protected EventType eventType;
protected Server server;
// sequence id generator for default FIFO ordering of events
protected final static AtomicLong seqids = new AtomicLong(0);
// sequence id for this event
private final long seqid;
// Listener to call pre- and post- processing. May be null.
private EventHandlerListener listener;
// Time to wait for events to happen, should be kept short
protected final int waitingTimeForEvents;
private final Span parent;
/**
* This interface provides pre- and post-process hooks for events.
*/
public interface EventHandlerListener {
/**
* Called before any event is processed
*
* @param event
* The event handler whose process method is about to be called.
*/
public void beforeProcess(EventHandler event);
/**
* Called after any event is processed
*
* @param event
* The event handler whose process method is about to be called.
*/
public void afterProcess(EventHandler event);
}
/**
* List of all wasp event handler types. Event types are named by a
* convention: event type names specify the component from which the event
* originated and then where its destined -- e.g. FSERVER2ZK_ prefix means the
* event came from a fServer destined for zookeeper -- and then what the even
* is; e.g. ENTITYGROUP_OPENING.
*
* <p>
* We give the enums indices so we can add types later and keep them grouped
* together rather than have to add them always to the end as we would have to
* if we used raw enum ordinals.
*/
public enum EventType {
// Messages originating from FServer (NOTE: there is NO direct communication from
// FServer to FMaster). These are a result of FServer updates into ZK.
FSERVER_ZK_ENTITYGROUP_CLOSED (1, ExecutorType.MASTER_CLOSE_ENTITYGROUP), // FServer has finished closing a entityGroup
FSERVER_ZK_ENTITYGROUP_OPENING (2, null), // FServer is in process of opening a entityGroup
FSERVER_ZK_ENTITYGROUP_OPENED (3, ExecutorType.MASTER_OPEN_ENTITYGROUP), // FServer has finished opening a entityGroup
FSERVER_ZK_ENTITYGROUP_FAILED_OPEN (4, ExecutorType.MASTER_CLOSE_ENTITYGROUP), // FServer failed to open a entityGroup
FSERVER_ZK_ENTITYGROUP_SPLITTING (5, null), // RS has started a entityGroup split
FSERVER_ZK_ENTITYGROUP_SPLIT (6, ExecutorType.MASTER_SERVER_OPERATIONS), // RS split has completed.
// Messages originating from Master to FServer
M_FSERVER_OPEN_ENTITYGROUP (20, ExecutorType.FSERVER_OPEN_ENTITYGROUP), // Master asking FServer to open a entityGroup
M_FSERVER_CLOSE_ENTITYGROUP (21, ExecutorType.FSERVER_CLOSE_ENTITYGROUP), // Master asking FServer to close a entityGroup
// Messages originating from Client to Master
C_M_DELETE_TABLE (40, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to delete a table
C_M_DISABLE_TABLE (41, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to disable a table
C_M_ENABLE_TABLE (42, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to enable a table
C_M_MODIFY_TABLE (43, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to modify a table
C_M_ADD_COLUMN (44, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to add column to table
C_M_DELETE_COLUMN (45, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to delete column of table
C_M_MODIFY_COLUMN (46, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to modify column of table
C_M_CREATE_TABLE (47, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to create a table
C_M_CREATE_INDEX (48, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to create a index
C_M_DELETE_INDEX (49, ExecutorType.MASTER_TABLE_OPERATIONS), // Client asking Master to drop a index
C_M_TRUNCATE_TABLE(50, ExecutorType.MASTER_TABLE_OPERATIONS),
// Client asking Master to truncate a table
// Updates from master to ZK. This is done by the master and there is
// nothing to process by either Master or FServer
M_ZK_ENTITYGROUP_OFFLINE (60, null), // Master adds this entityGroup as offline in ZK
M_ZK_ENTITYGROUP_CLOSING (61, null), // Master adds this entityGroup as closing in ZK
// Master controlled events to be executed on the master
M_SERVER_SHUTDOWN (80, ExecutorType.MASTER_SERVER_OPERATIONS); // Master is processing shutdown of a FServer
private final int code;
private final ExecutorService.ExecutorType executor;
/**
* Constructor
*/
EventType(final int code, final ExecutorType executor) {
this.code = code;
this.executor = executor;
}
public int getCode() {
return this.code;
}
public static EventType get(final int code) {
// Is this going to be slow? Its used rare but still...
for (EventType et : EventType.values()) {
if (et.getCode() == code)
return et;
}
throw new IllegalArgumentException("Unknown code " + code);
}
public boolean isOnlineSchemaChangeSupported() {
return (this.equals(EventType.C_M_MODIFY_TABLE));
}
ExecutorType getExecutorServiceType() {
return this.executor;
}
}
/**
* Default base class constructor.
*/
public EventHandler(Server server, EventType eventType) {
this.parent = Trace.currentTrace();
this.server = server;
this.eventType = eventType;
seqid = seqids.incrementAndGet();
this.waitingTimeForEvents = server.getConfiguration().getInt(
"wasp.master.event.waiting.time", 1000);
}
public void run() {
Span chunk = Trace.startSpan(Thread.currentThread().getName(), parent,
Sampler.ALWAYS);
try {
if (getListener() != null)
getListener().beforeProcess(this);
process();
if (getListener() != null)
getListener().afterProcess(this);
} catch (Throwable t) {
LOG.error("Caught throwable while processing event " + eventType, t);
} finally {
chunk.stop();
}
}
/**
* This method is the main processing loop to be implemented by the various
* subclasses.
*
* @throws java.io.IOException
*/
public abstract void process() throws IOException;
/**
* Return the event type
*
* @return The event type.
*/
public EventType getEventType() {
return this.eventType;
}
/**
* Get the priority level for this handler instance. This uses natural
* ordering so lower numbers are higher priority.
* <p>
* Lowest priority is Integer.MAX_VALUE. Highest priority is 0.
* <p>
* Subclasses should override this method to allow prioritizing handlers.
* <p>
* Handlers with the same priority are handled in FIFO order.
* <p>
*
* @return Integer.MAX_VALUE by default, override to set higher priorities
*/
public int getPriority() {
return Integer.MAX_VALUE;
}
/**
* @return This events' sequence id.
*/
public long getSeqid() {
return this.seqid;
}
/**
* Default prioritized runnable comparator which implements a FIFO ordering.
* <p>
* Subclasses should not override this. Instead, if they want to implement
* priority beyond FIFO, they should override {@link #getPriority()}.
*/
@Override
public int compareTo(Runnable o) {
EventHandler eh = (EventHandler) o;
if (getPriority() != eh.getPriority()) {
return (getPriority() < eh.getPriority()) ? -1 : 1;
}
return (this.seqid < eh.seqid) ? -1 : 1;
}
/**
* @return Current listener or null if none set.
*/
public synchronized EventHandlerListener getListener() {
return listener;
}
/**
* @param listener
* Listener to call pre- and post- {@link #process()}.
*/
public synchronized void setListener(EventHandlerListener listener) {
this.listener = listener;
}
@Override
public String toString() {
return "Event #" + getSeqid() + " of type " + eventType + " ("
+ getInformativeName() + ")";
}
/**
* Event implementations should override thie class to provide an informative
* name about what event they are handling. For example, event-specific
* information such as which entityGroup or server is being processed should
* be included if possible.
*/
public String getInformativeName() {
return this.getClass().toString();
}
}