Plugin mechanism for messaging layer (Issue #372)

[anfeng]

This pull request proposes a plugin mechanism for messaging layer.

We are introducing a new Storm config parameter, storm.messaging.transport, to specify a desired plugin. For now, defaults.yaml states that zmq is such a transport.

A messaging plugin needs to implements a simple interface, ITransport, and should have a default constructor and implements, newContext(), to return an IContext.
public interface ITransport {
public IContext newContext();
}

IContext is defined to have the following methods, where prepare(storm_conf) is invoked immediately after IContext is constructed by Storm core (TransportFactory).
public interface IContext {
public void prepare(Map storm_conf);

public IConnection bind(String storm_id, int port);
public IConnection connect(String storm_id, String host, int port);
public void term();

};

IConnection is defined as below, where TaskMessage is a simple Java class contains task ID and message.
public interface IConnection {
public TaskMessage recv();
public TaskMessage recv_with_flags(int flags);
public void send(int task, byte[] message);
public void close();
}

I have modified zmq.clj to implements all these interfaces. local.clj has been revised to implement IContext and IConnection.

worker.clj has been revised slightly to leverage the plugin mechanism. It does not knows ZMQ anymore :-)

[anfeng]

We have now 2 messaging plugins: local.TransportPlugin and zmq.TransportPlugin. The newly created test program illustrates how topologies should work with such plugins.

[nathanmarz]

This test doesn't test the messaging as there's only one worker.

[nathanmarz]

A few additional notes besides the inline comments I made:

1. All throughout you use Java interop except what you're interopping against isn't type-hinted. You have to use type hints whenever you use Java interop or else the compiler will use Java reflection (which is incredibly slow). To add a type hint, just preface the relavant portion with ^{type}, e.g.:

(.foo ^MyType (:some-object some-map) ...)

1. Looks like this pull request conflicts with some changes I pushed last night. Can you fix the conflicts?

[anfeng]

Nathan:

I have made the following changes with the latest commit:
(1) messaging_test.clj now uses zmq plugin
(2) merged with your recent changes at nathanmaz/storm master
(3) added type hints

I could not remove ITransport due to the limitation of Clojure deftype. For both zmq and local tranport, we use clojure to define a context class, which needs to have some data fields. Clojure deftypes requires all those fields are be defined and passed in via constructor parameters. Clojure does not allow us to create a default constructor in such cases. Any suggestion?

[anfeng]

Nathan:

I have addressed all issues you raised so far.
(1) Interface classes simplified with IContext and IConnextion
(2) messaging_test.clj now uses zmq plugin
(3) merged with your recent changes at nathanmaz/storm master
(4) added type hints

zmq plugin is now implemented via Java, not Clojure.

[anfeng]

I created a simple implementation of this plugin API based on Java NIO.2. It's available at https://github.com/anfeng/storm-messaging-nio

Once you accept this pull request, I will work on NIO.2 and/or Netty implementation. NIO.2 is simpler than Netty, but requires dependency on Java7.

[nathanmarz]

This needs a type hint

[nathanmarz]

Let's actually keep the zmq stuff implemented in clojure and do the zmq -> java stuff in a separate pull request. Then I'm comfortable merging this in.

[anfeng]

Please take a look at this revision. zmq is now put back into Clojure.

TransportFactory has a slightly modified logic.
(1) If the specified plugin java class implements IContext, we create its instance and invoke .prepare.
(2) If the specified plugin java class doesnot impelment IContext, it should have a method makeContext(storm_conf) that returns an IContext object.

[nathanmarz]

Let's get rid of these log-debugs in the critical path. They actually show up in the profiler.

[nathanmarz]

Made a few more comments. As soon as those are addressed I can merge this in.

[anfeng]

Code revised per your suggestion. Hopefully we are ready to merge.

[nathanmarz]

Merged.