org.eclipse.smila.taskmanager.persistence.zk

Class ZkTaskQueue

java.lang.Object

org.eclipse.smila.taskmanager.persistence.zk.ZkTaskQueue

public class ZkTaskQueue
extends java.lang.Object

Queue implementation wrapping a zookeeper client. Tasks are stored in different (virtual) sub queues identified by task property. For those sub queues, tasks are delivered round robin (-> fairness). Does not do ZooKeeper session handling, it must be done from the calling context. Zookeeper structure: (task-id nodes contain Task as data) /smila/tasks//todo/<sub-queue>/<counter>-<task-id>-<host name> ........................../todo_qualified/<sub-queue>/<qualifier-string>/<counter>-<task-id>-<host name> ........................../inprogress/ ........................../delivery/<sub-queue>/<worker-name>/<delivery-time>

Field Summary

Fields

Modifier and Type	Field and Description
static java.lang.String	DEFAULT_SUB_QUEUE use this as a default for identifying the right (sub) queue if no property is given in the task.
static java.lang.String	DELIVERY_PREFIX zk path for delayed task delivery data.
static java.lang.String	INPROGRESSDIR_SUFFIX prefix for task-in-progress directories.
static java.lang.String	OWNER_PROP Property identifying owner of this task.
static java.lang.String	ROOT_DIR_PREFIX prefix for task directories.
static java.lang.String	SECTION_INPROGRESS Property for task-in-progress.
static java.lang.String	SECTION_TODO Property for tasks-todo.
static java.lang.String	SECTION_TODOQUALIFIED Property for qualified-tasks-todo.
static java.lang.String	TASKDIR_PREFIX prefix for task directories.
static java.lang.String	TODODIR_SUFFIX prefix for tasks-todo directories.
static java.lang.String	TODOQUALIFIEDDIR_SUFFIX prefix for qualified-tasks-todo directories.

Constructor Summary

Constructors

Constructor and Description
ZkTaskQueue(ZooKeeperService service, java.lang.String workerName, java.lang.String localhost)
ZkTaskQueue(ZooKeeperService service, java.lang.String workerName, java.lang.String localhost, long maxNoOfTasksPerHost)

Method Summary

Methods

Modifier and Type	Method and Description
void	cleanEmptyNodes(long timeToLiveMs) Remove empty task nodes in the todo(_qualified) directories (leftovers from failed create operations) empty todo_qualified directories that seem not to be in use anymore.
void	delete(java.lang.String taskId) Deletes the given task from the inprogress section.
void	disconnectZkSession() for testing only!
java.util.List<Task>	filterDuplicates(java.util.List<Task> tasks) return new list of tasks with only those tasks from the given task list that do not have property Task.PROPERTY_UNIQUENESS_TAG set or for which currently no task is in the Todo queue that has the same uniqueness tag.
Task	get(java.util.Collection<java.lang.String> qualifiers, java.lang.String host) Return next task that can be processed.
Task	getInProgressTask(java.lang.String taskId) read the task content of an in-progress task.
long	getMaxNoOfTasksPerHost()
TaskCounter	getTaskCounter()
AnyMap	getTaskInfo(java.lang.String section, java.lang.String taskName) Prepares information on a task stored in the task storage.
TaskList	getTaskList(java.lang.String section, int maxCount) Get task list for current task pipe and sub-type.
java.util.Collection<java.lang.String>	getTimedOutTasks(long timeToLiveMs)
java.lang.String	getWorkerName() return worker name.
boolean	hasQualifiedTasks()
void	keepAlive(java.lang.String taskId) Process isAlive call for given task.
void	purge() Remove all task (and todo sub queue) nodes.
void	put(Task task) Puts a task in our task pipe.
java.lang.String	putInProgress(Task task) write task to in-progress section of the queue.
void	removeTasks(AnyMap filterMap) Remove canceled tasks identified by the filter map.
void	setMaxNoOfTasksPerHost(long maxNoOfTasksPerHost) set new scale-up limit.

Methods inherited from class java.lang.Object

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

Field Detail

ROOT_DIR_PREFIX

public static final java.lang.String ROOT_DIR_PREFIX

prefix for task directories.

See Also:

Constant Field Values

TASKDIR_PREFIX

public static final java.lang.String TASKDIR_PREFIX

prefix for task directories.

See Also:

Constant Field Values

TODODIR_SUFFIX

public static final java.lang.String TODODIR_SUFFIX

prefix for tasks-todo directories.

See Also:

Constant Field Values

TODOQUALIFIEDDIR_SUFFIX

public static final java.lang.String TODOQUALIFIEDDIR_SUFFIX

prefix for qualified-tasks-todo directories.

See Also:

Constant Field Values

INPROGRESSDIR_SUFFIX

public static final java.lang.String INPROGRESSDIR_SUFFIX

prefix for task-in-progress directories.

See Also:

Constant Field Values

DELIVERY_PREFIX

public static final java.lang.String DELIVERY_PREFIX

zk path for delayed task delivery data.

See Also:

Constant Field Values

OWNER_PROP

public static final java.lang.String OWNER_PROP

Property identifying owner of this task.

See Also:

Constant Field Values

SECTION_TODO

public static final java.lang.String SECTION_TODO

Property for tasks-todo.

See Also:

Constant Field Values

SECTION_TODOQUALIFIED

public static final java.lang.String SECTION_TODOQUALIFIED

Property for qualified-tasks-todo.

See Also:

Constant Field Values

SECTION_INPROGRESS

public static final java.lang.String SECTION_INPROGRESS

Property for task-in-progress.

See Also:

Constant Field Values

DEFAULT_SUB_QUEUE

public static final java.lang.String DEFAULT_SUB_QUEUE

use this as a default for identifying the right (sub) queue if no property is given in the task.

See Also:

Constant Field Values

Constructor Detail

ZkTaskQueue

public ZkTaskQueue(ZooKeeperService service,
           java.lang.String workerName,
           java.lang.String localhost)

Parameters:

service - zookeeper service needed to get zookeeper client

workerName - name of worker for which we are managing tasks

localhost - local host name

ZkTaskQueue

public ZkTaskQueue(ZooKeeperService service,
           java.lang.String workerName,
           java.lang.String localhost,
           long maxNoOfTasksPerHost)

Parameters:

service - zookeeper service needed to get zookeeper client

workerName - name of worker for which we are managing tasks

localhost - local host name

maxNoOfTasksPerHost -

Method Detail

disconnectZkSession

public void disconnectZkSession()
                         throws TaskmanagerException

for testing only!

Throws:

TaskmanagerException

getMaxNoOfTasksPerHost

public long getMaxNoOfTasksPerHost()

Returns:

scale-up limit setting.

setMaxNoOfTasksPerHost

public void setMaxNoOfTasksPerHost(long maxNoOfTasksPerHost)

set new scale-up limit.

getWorkerName

public java.lang.String getWorkerName()

return worker name.

put

public void put(Task task)
         throws TaskmanagerException

Puts a task in our task pipe.

Parameters:

task - the task

Throws:

TaskmanagerException - error

putInProgress

public java.lang.String putInProgress(Task task)
                               throws TaskmanagerException

write task to in-progress section of the queue.

Parameters:

task - task

Returns:

value of OWNER_PROP property.

Throws:

TaskmanagerException

get

public Task get(java.util.Collection<java.lang.String> qualifiers,
       java.lang.String host)
         throws TaskmanagerException

Return next task that can be processed. This method is not thread-safe, it must be called in a synchronized context!

Parameters:

qualifiers - optional task qualifiers

host - host name where the worker is running that requested the task, to be stored as property in the task

Returns:

next task that can be processed.

Throws:

TaskmanagerException - error

getInProgressTask

public Task getInProgressTask(java.lang.String taskId)
                       throws TaskmanagerException

read the task content of an in-progress task. The task is not deleted yet, but kept alive to prevent concurrent rollback.

Parameters:

taskId - task Id

Returns:

task contents or null if no such task exists or is already finishing

Throws:

TaskmanagerException - if task is not in-progress or other errors.

delete

public void delete(java.lang.String taskId)
            throws TaskmanagerException

Deletes the given task from the inprogress section.

Parameters:

taskId - identifies the task

Throws:

TaskmanagerException - error; BadParameterTaskmanagerException if taskId isn't found

keepAlive

public void keepAlive(java.lang.String taskId)
               throws TaskmanagerException

Process isAlive call for given task.

Parameters:

taskId - identifies the task

Throws:

TaskmanagerException - error; BadParameterTaskmanagerException if taskId isn't found

hasQualifiedTasks

public boolean hasQualifiedTasks()
                          throws KeeperException

Returns:

true if this queue has a directory for qualified tasks and it contains at least one sub-queue node with qualifier(s).

Throws:

KeeperException

getTaskCounter

public TaskCounter getTaskCounter()
                           throws TaskmanagerException

Returns:

task counter

Throws:

TaskmanagerException - error

getTimedOutTasks

public java.util.Collection<java.lang.String> getTimedOutTasks(long timeToLiveMs)

Parameters:

timeToLiveMs - the maximum processing time

Returns:

task Ids of tasks that have exceeded the timeToLive.

purge

public void purge()

Remove all task (and todo sub queue) nodes. Errors are logged as warning, but the purge tries to continue.

cleanEmptyNodes

public void cleanEmptyNodes(long timeToLiveMs)

Remove

• empty task nodes in the todo(_qualified) directories (leftovers from failed create operations)
• empty todo_qualified directories that seem not to be in use anymore.

Errors are logged as warning, but the clean tries to continue.

Parameters:

timeToLiveMs - time in milliseconds after which to clean up stale tasks or empty todo_qualified dirs.

getTaskList

public TaskList getTaskList(java.lang.String section,
                   int maxCount)
                     throws TaskmanagerException

Get task list for current task pipe and sub-type.

Parameters:

section - task pipe sub-type: "inprogress" or "todo".

maxCount - max. number of tasks in return list

Returns:

task list

Throws:

TaskmanagerException - error

getTaskInfo

public AnyMap getTaskInfo(java.lang.String section,
                 java.lang.String taskName)
                   throws TaskmanagerException

Prepares information on a task stored in the task storage.

Parameters:

section - queue section ("todo" or "inprogress")

taskName - task identifier

Returns:

task information

Throws:

TaskmanagerException - on error

removeTasks

public void removeTasks(AnyMap filterMap)

Remove canceled tasks identified by the filter map.

Parameters:

filterMap - map to identify tasks to be removed

filterDuplicates

public java.util.List<Task> filterDuplicates(java.util.List<Task> tasks)
                                      throws TaskmanagerException

return new list of tasks with only those tasks from the given task list

• that do not have property Task.PROPERTY_UNIQUENESS_TAG set or
• for which currently no task is in the Todo queue that has the same uniqueness tag.

Parameters:

tasks - to filter for this task queue

Returns:

tasks that are not already contained in current task queue's todo list

Throws:

TaskmanagerException