org.apache.oozie.service

Class ZKJobsConcurrencyService

java.lang.Object

org.apache.oozie.service.JobsConcurrencyService

org.apache.oozie.service.ZKJobsConcurrencyService

All Implemented Interfaces:

Service, Instrumentable

public class ZKJobsConcurrencyService
extends JobsConcurrencyService
implements Service, Instrumentable

This Service helps coordinate other Services to prevent duplicate processing of Jobs if there are multiple Oozie Servers. This implementation uses ZooKeeper and is designed to correctly deal with multiple Oozie Servers.

The distributed locks provided by ZKLocksService will prevent any concurrency issues from occurring if multiple Oozie Servers try to process the same job at the same time. However, this will make Oozie slower (more waiting on locks) and will place additional stress on ZooKeeper and the Database. By "assigning" different Oozie servers to process different jobs, we can improve this situation. This is particularly necessary for Services like the RecoveryService, which could duplicate jobs otherwise. We can assign jobs to servers by doing a mod of the jobs' id and the number of servers.

The leader server is elected by all of the Oozie servers, so there can only be one at a time. This is useful for tasks that require (or are better off) being done by only one server (e.g. database purging). Note that the leader server isn't a "traditional leader" in the sense that it doesn't command or have authority over the other servers. This leader election uses a znode under /oozie.zookeeper.namespace/ZK_BASE_SERVICES_PATH/ZK_LEADER_PATH (default is /oozie/services/concurrencyleader).

Field Summary

Fields inherited from interface org.apache.oozie.service.Service

CONF_PREFIX, DEFAULT_LOCK_TIMEOUT, lockTimeout

Constructor Summary

Constructors

Constructor and Description
ZKJobsConcurrencyService()

Method Summary

Methods

Modifier and Type	Method and Description
void	destroy() Destroy the zookeeper jobs concurrency service.
List<String>	getJobIdsForThisServer(List<String> ids) Filter out any job ids that should not be processed by this server.
Map<String,String>	getOtherServerUrls() Return a map of instance id to Oozie server URL of other servers.
Map<String,String>	getServerUrls() Return a map of instance id to Oozie server URL.
void	init(Services services) Initialize the zookeeper jobs concurrency service
void	instrument(Instrumentation instr) Instruments the zk jobs concurrency service.
boolean	isAllServerRequest(Map<String,String[]> params) Checks if rest request is for all server.
boolean	isHighlyAvailableMode() Return if it is running in HA mode
boolean	isJobIdForThisServer(String jobId) Check to see if jobId should be processed by this server.
boolean	isLeader() Check to see if this server is the leader server.

Methods inherited from class org.apache.oozie.service.JobsConcurrencyService

getInterface

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.oozie.service.Service

getInterface

Constructor Detail

ZKJobsConcurrencyService

public ZKJobsConcurrencyService()

Method Detail

init

public void init(Services services)
          throws ServiceException

Initialize the zookeeper jobs concurrency service

Specified by:

init in interface Service

Overrides:

init in class JobsConcurrencyService

Parameters:

services - services instance.

Throws:

ServiceException - thrown if the service could not initialize.

destroy

public void destroy()

Destroy the zookeeper jobs concurrency service.

Specified by:

destroy in interface Service

Overrides:

destroy in class JobsConcurrencyService

instrument

public void instrument(Instrumentation instr)

Instruments the zk jobs concurrency service.

Specified by:

instrument in interface Instrumentable

Overrides:

instrument in class JobsConcurrencyService

Parameters:

instr - instance to instrument the zookeeper jobs concurrency service to.

isLeader

public boolean isLeader()

Check to see if this server is the leader server. This implementation only returns true if this server has been elected by all of the servers as the leader server.

Overrides:

isLeader in class JobsConcurrencyService

Returns:

true if this server is the leader; false if not

isJobIdForThisServer

public boolean isJobIdForThisServer(String jobId)

Check to see if jobId should be processed by this server. This implementation only returns true if the index of this server in ZooKeeper's list of servers is equal to the id of the job mod the number of servers.

Overrides:

isJobIdForThisServer in class JobsConcurrencyService

Parameters:

jobId - The jobId to check

Returns:

true if this server should process this jobId; false if not

getJobIdsForThisServer

public List<String> getJobIdsForThisServer(List<String> ids)

Filter out any job ids that should not be processed by this server. This implementation only preserves jobs such that the index of this server in ZooKeeper's list of servers is equal to the id of the job mod the number of servers.

Overrides:

getJobIdsForThisServer in class JobsConcurrencyService

Parameters:

ids - The list of job ids to check

Returns:

a filtered list of job ids that this server should process

getServerUrls

public Map<String,String> getServerUrls()

Return a map of instance id to Oozie server URL. This implementation always returns a map with where the key is the instance id and the value is the URL of each Oozie server that we can see in the service discovery in ZooKeeper.

Overrides:

getServerUrls in class JobsConcurrencyService

Returns:

A map of Oozie instance ids and URLs

getOtherServerUrls

public Map<String,String> getOtherServerUrls()

Return a map of instance id to Oozie server URL of other servers. This implementation always returns a map with where the key is the instance id and the value is the URL of each Oozie server that we can see in the service discovery in ZooKeeper.

Overrides:

getOtherServerUrls in class JobsConcurrencyService

Returns:

A map of Oozie instance ids and URLs

isAllServerRequest

public boolean isAllServerRequest(Map<String,String[]> params)

Checks if rest request is for all server. By default it's true.

Overrides:

isAllServerRequest in class JobsConcurrencyService

Parameters:

params - the HttpRequest param

Returns:

false if allservers=false, else true;

isHighlyAvailableMode

public boolean isHighlyAvailableMode()

Return if it is running in HA mode

Overrides:

isHighlyAvailableMode in class JobsConcurrencyService

Returns: