[ http://issues.apache.org/jira/browse/NUTCH-368?page=comments#action_12435549 ] Andrzej Bialecki commented on NUTCH-368: -----------------------------------------
It is modeled after the core concepts in JMS, in the sense that there are topics, queues and messages. Of course it's a simplification, but there are many similarities, so for people familiar with JMS it should also look familiar. Highlights of JMS vs. this API: * this API works now ;) I haven't even looked at implementing a JMS-compliant provider on top of Hadoop, it would be surely much more complicated and take much more time... * in JMS there are point-to-point and publish-subscribe communication models. This API is most similar to the pub-sub model, in the sense that each message consumer tracks messages it consumed independently of all other consumers. I selected this model in order to support sending messages to map-reduce jobs, where there are many receivers (tasks) and possibly many senders. * the API to browse topics, and select which messages to process (in-order or out-of-order) is IMHO much more natural than the one in JMS API. * JMS provides PERSISTENT and NON_PERSISTENT delivery modes. This API provides only a persistent mode; however, at the end of the job applications may elect to remove all queues and their content. * JMS supports both blocking and non-blocking reads. This API only supports non-blocking reads. * both JMS and this API support event-driven message processing. * JMS supports "message selectors" which filter incoming messages according to their properties. This API doesn't support it, but it can be easily added if needed. * JMS supports Publishers and Subscribers as separate interfaces. This API doesn't make such distinction - each client connected to a queue may both receive and send messages. * JMS supports transaction-oriented communication, this is not supported here. * the only persistency mechanism supported now is plain files on Hadoop FileSystem (either local or DFS), and it works transparently, i.e. it doesn't require any special support from Hadoop. * messages may contain any Writable, so it's a natural API from the Hadoop's point of view. So IMHO this gives a fairly large subset of JMS functionality in a simple to understand (and maintain) implementation. Additionally, it doesn't require any modifications in Hadoop, although it could surely use some, to better integrate with map-reduce jobs - e.g. TaskTrackers could be responsible for starting queue sessions for jobs that indicate this; instead of polling for FileSystem updates we could have filesystem monitors, etc ... but this is not strictly necessary, this API works as it is now. > Message queueing system > ----------------------- > > Key: NUTCH-368 > URL: http://issues.apache.org/jira/browse/NUTCH-368 > Project: Nutch > Issue Type: New Feature > Affects Versions: 0.9.0 > Reporter: Andrzej Bialecki > Assigned To: Andrzej Bialecki > Attachments: msg.tgz > > > This is an implementation of a filesystem-based message queueing system. The > motivation for this functionality is explained in HADOOP-490 - there is > nothing Nutch-specific in this implementation, so if it's considered > generally useful it could be moved there. > Below are excerpts from the included javadocs. > The model of the system is as follows: > * applications (including map-reduce jobs) may create their own separate > message queueing area. Alternatively, they can specifically ask for a named > message queue, belonging to a different application or existing as a > system-wide queue. Message queues are created under "/mq" and then the > message queue id (for map-reduce jobs this is a job id, or it can be any > other name passed as job id to the constructor). > Please see the example for more information. > * a single unit of information passing through queues is a Msg, which has > a unique identifier (consisting of creation time and publisher name), string > subject, and content (Writable). > * single MsgQueue in fact consists of any number of topics. There are > four predefined ones: in, out, err, and ctrl. > * messages are published to topics, which present a sequential view of > messages, sorted by msgId (which corresponds to their order of arrival). > * each message queue may periodically poll for changes > (MsgQueue.startPolling()), using a separate thread. Polling updates the list > of topics and messages. Poll interval is configurable, and defaults to 5 sec. > * each detected change in the queue (add/remove topic, add/remove > message) may be communicated to registered listeners. Out-of-band messages > are not supported in this version, but it's not too complicated to add them. > Applications can create listeners watching queues for newly added messages, > or deleted messages, added topics or deleted topics, etc. > * each instance of MsgQueue using the same physical queue maintains its > own view of the queue, keeping track of topics and messages that it considers > "processed and discarded". In other words, multiple readers and creators may > modify queues, and each knows which messages it already processed and which > ones are new. In a similar fashion, instances may willfully "remove" certain > topics from their view, even though these topics still physically exist and > are available for other instances (and later on they can "add" them to their > view again). > This somewhat complicated feature was implemented in order to support > multiple readers for the same message (e.g. many tasks per one mapred job). > Each task needs to register for the same queue, and if they didn't have their > own views of the queue, messages would be consumed by the first task that got > to them. As it is implemented now, each task may consume messages at its own > pace. At the end of the job applications may elect to keep the queue around > or to destroy it (and thus remove all topics and messages in it). > * messages, topics and queues may be destroyed by any user, at which > point they are physically removed from the filesystem. All users will > gradually update their views, during the next poll operation. > * there is a command-line tool to examine and modify queues, and also to > retrieve and send simple text messages. You can run it like this: > bin/nutch org.apache.nutch.util.msg.MsgQueueTool ...many options... -- This message is automatically generated by JIRA. - If you think it was sent incorrectly contact one of the administrators: http://issues.apache.org/jira/secure/Administrators.jspa - For more information on JIRA, see: http://www.atlassian.com/software/jira
