Google Code, Git, and Eclipse

Create a new Project in Google Code

• Open any projects.
• From "My favorites" drop-down menu, you will see "Create a project...", click on it.
• Fill in all Options except Project labels. Choose Git as the Version Control System.

• Click on "Create project" button.

Find Access Information

• Open "Source" tab.
• etc. etc. etc.

Clone a Repository

• Run this command

git clone https://<your user name>@code.google.com/p/simple-demo-set/ 

• Create a RADME.txt under the repository.
• git add README.txt
• git commit -m "Add the first file"
• create $HOME/.netrc (I'm using cygwin)

machine code.google.com

login XXXX

password GOOGLEGENERATED

• git push -u origin master

Add Git Repository into Eclipse

• Open Git Repositories Exploring Perspective
• Click on "Add an existing local Git Repository to this view"
• Select the designated directory location and clikc on Finish.

Share Project from Eclipse

http://www.vogella.com/articles/EGit/article.html#usingegit_shareproject

• Choose the repository instead of create a local one.

Create .gitignore

It seems that Eclipse's global ignorance settings don't work for Git. You have to create your own .gitignore for the git repository. These are the entry you probably want to add.

.classpath

.project

.settings

target

Storm?

• Storm, distributed and fault-tolerant realtime computation
• Twitter Storm: Open Source Real-time Hadoop
• nathanmarz/storm · GitHub
• Open source Java projects: Storm - JavaWorld

• Whereas Hadoop targets batch processing, Storm is an always-active service that receives and processes unbound streams of data.
• Like Hadoop, Storm is a distributed system that offers massive scalability for applications that store and manipulate big data. 
• Unlike Hadoop, it delivers that data instantaneously, in realtime.
• It is written primarily in Clojure and supports Java by default.
• Use cases
• Realtime analytics
• Online machine learning
• Continuous computation
• Distributed RPC
• ETL
• How does storm differ from Hadoop?
• The simple answer is that Storm analyzes realtime data while Hadoop analyze offline data.
• In truth, the two frameworks complement one another more than they compete.
• Hadoop
• Provides its own file system (HDFS)
• Manages both data and code/tasks.
• It divides data into blocks and when a "job" executes, it pushes analysis code close to the data it is analyzing.
• This is how Hadoop avoids the overhead of network communication in loading data -- keeping the analysis code next to the data enables Hadoop to read it faster by orders of magnitude.
• MapReduce
• Hadoop partitions data into chunks and passes those chunks to mappers that map keys to values.
• Reducers then assemble those mapped key/value pairs into a usable output.
• The MapReduce paradigm operates quite elegantly but is targeted at data analysis.
• HDFS
• In order to leverage all the power of Hadoop application data must be stored in the HDFS file system.
• Storm
• Storm solves a different problem altogether.
• Realtime
• meaning right now
• Storm is interested in understanding things that are happening in realtime, and interpreting them.
• File System
• Storm does not have its own file system.
• Programming Paradigm
• Its programming paradigm is quite a bit different from Hadoop's.
• Storm is all about obtaining chunks of data, known as spouts, from somewhere and passing that data through various processing components, known as bolts.
• Storm's data processing mechanism is extremely fast and is meant to help you identify live trends as they are happening.
• Unlike Hadoop, Storm doesn't care what happened yesterday or last week.
• Architecture
• At the highest level, Storm is comprised of topologies.
• A topology is a graph of computations
• Each node contains processing logic and each path between nodes indicates how data should be passed between nodes.
• In side of topologies you have networks of streams, which are unbounded sequences of tuples.
• Storm provides a mechanism to transform streams into new streams using spouts and bolts.
• Spouts
• Spouts generate streams, which can pull data from a site like Twitter of Facebook and then publish it in an abstract format.
• Bolts
• Bolts consume input streams, process them, and then optionally generate new streams.
• Tuples
• Storm's data model is represented by tuples.
• A tuples is a named list of values of any type.
• Storm supports all primititve types, Strings, and byte-arrays and you can build your own serializer if you want to use your own object types.
• Your spouts will "emit" tuples 
• And your bolts will consume them.
• Your bolts may also emit tuples if their output is destined to be processed by another bolt downstream.
• Basically, emitting tuples is the mechanism for passing data from a spout to a bolt, or from a bolt to another bolt.

This is quite a normal architecture for expandable or scalable computatal system, or network. And basically, the name, Topologies, the graph, makes a lot of sense.

It's an Oriented Graph. If we add a root node above Spouts, it would look like a tree but with shared children. It reminds me the Traveling Saleman Problem (TSP, Travelling salesman problem - Wikipedia, the free encyclopedia).

• Storm Cluster
• A Storm Cluster is somewhat similar to Hadoop clusters, but while a Hadoop cluster runs map-reduce jobs, Storm runs topologies.
• Map-reduce jobs eventually end.
• Topologies are destined to run until you explicitly kill them.
• Storm clusters define two types of nodes
• Master Node
• This node runs a daemon process called Nimbus.
• Nimbus is responsible for distributing code across the cluster, assigning tasks to machines, and monitoring the success and failure of units of work.
• Worker Nodes
• These nodes run a daemon process called the Supervisor.
• A Supervisor is responsible for listening for work assignments for its machine.
• It then subsequently starts and stops worker processes. Each worker process executes a subset of a topology, so that the execution of a topology is spread across a multitude of worker processes running on a multitude of machines.

This is a typical cluster architecture. I used to designed a system with the Master Node named MC (MultiController), and the Worker Nodes named LC (LocalController). MC takes care of job distribution. But MC doesn't monitor the success or failure of units of work. It monitors the LC's health and relocate the Work Node to ensure High Availability.

My system was not designed for dynamic expansion. There are only three fixed layers. LC, the Worker Node runs as a daemon and supervises a list of Engines. Those Engines are created dynamically by some configuration package, the work assignments if you will. It starts and stops these Engines, the Worker Processes if you will. Those Worker Processes work basically independently, which made my cluster simpler then this. I think having each worker process executes a subset of a topology, and coordinates those topology will be a challenge.

In my cluster, each Worker Nodes takes care of the success or failure of its own Worker Processes.

• ZooKeeper
• Sitting between the Nimbus and the various Supervisors is ZooKeeper.
• ZooKeeper's goal is to enable highly reliable distributed coordination, mainly by acting as a centralized service for distributed cluster functionality.
• Storm topologies are deployed to the Nimbus and then the Nimbus deploys spouts and bolts to Supervisors.
• When it comes time to execute spouts and bolts, the Nimbus communicates with the Supervisors by passing messages to ZooKeepers.
• Zookeepers maintain all state for the topology, which allows the Nimbus and Supervisors to be fail-fast and stateless: if the Nimbus or Supervisor processes go down then the state of processing is not lost;
• If the Nimbus or Supervisor processes go down then the state of processing is not lost;
• Work is reassigned to another Supervisor and processing continues.
• Then, when the Nimbus or a Supervisor is restarted, they simply rejoin the cluster and their capacity is added to the cluster.

My first question would be: how many Nimbus in the system? It seems only one. If this is true, we need somewhere to eliminate this single-failure-point (Single point of failure - Wikipedia, the free encyclopedia). So I think Zookeeper actions as an independent configuration storage out of Nimbus. Nimbus delegates Zookeepers to deployed Supervisors and manage them. Zookeepers are clustered and do not have single point of failure problem.

Of course, this is what I guess so far.

• Storm Development Environment
• Local Mode:
• Storm executes topologies completely in-process by simulating worker nodes using threads.
• Distributed Mode:
• In distributed mode, it runs across a cluster of machines.

Channel Channel Channel!!

DirectChannel

First of all, DirectChannel is not Pollable Channel. Since it is not Pollable Channel, it is not based on BlockingQueue. I was wrong about that before.

DirectChannel is a Subscribable Channel but with Point-to-point Semantics. It would be easy to understand from the implementation's point of view.

private boolean doDispatch(Message<?> message) {

boolean success = false;

Iterator<MessageHandler> handlerIterator = this.getHandlerIterator(message);

if (!handlerIterator.hasNext()) {

throw new MessageDispatchingException(message, "Dispatcher has no subscribers");

}

List<RuntimeException> exceptions = new ArrayList<RuntimeException>();

while (success == false && handlerIterator.hasNext()) {

MessageHandler handler = handlerIterator.next();

try {

handler.handleMessage(message);

success = true; // we have a winner.

}

catch (Exception e) {

RuntimeException runtimeException = (e instanceof RuntimeException)

? (RuntimeException) e

: new MessageDeliveryException(message,

"Dispatcher failed to deliver Message.", e);

if (e instanceof MessagingException &&

((MessagingException) e).getFailedMessage() == null) {

((MessagingException) e).setFailedMessage(message);

}

exceptions.add(runtimeException);

this.handleExceptions(exceptions, message, !handlerIterator.hasNext());

}

}

return success;

}

DirectChannel picks up the next available Handler and calls its handleMessage() method to send out the message. And that's it.

• It's point-to-point.
• It only call one handler successfully.
• And it won't queue!
• When the message arrives and the handler is not there, it will throw an exception. 

Half-duplex Mode

Slow network card: what is half duplex mode? | TuxRadar Linux
Change NIC settings betwee Full / Half duplex
Going from full to half duplex on cable, any real speed difference on Net? - AnandTech Forums

In full duplex operation, you have, in essence, 10Mb/s guaranteed bandwidth up and down. When you go to full duplex, you move back to "regular" ethernet, which is a collision-based system. In a half duplex network, it's a first-come-first-serve operation. Your computer transmits when it sees the network is clear. It is very possible (and, in fact, very very likely under high traffic conditions) that another computer will have ALSO started transmitting at the same time you did. Both your signals go across the wire and interfere with each other, effectively cancelling out both transmissions. This is called a collision. When your computer sees a collision it stops transmitting and pauses for a random amount of time before trying to re-transmit.

What is Half Duplex and Full Duplex Ethernet Modes? | Technology Updates

Ethernet IEEE 802.3 standard defines the half duplex; Cisco describes, it uses a digital signal on a wire pair flowing in both tracks on the wire. Half Duplex always employs the Carrier Sense Multiple Access with Collision Domain (CSMA/CD), so that it can retransmit the transmission if collision occurs. Hub always works in half duplex mode. Half duplex Ethernet is not efficient because it has the limit up to 10BaseT, as Cisco describes 10BaseT is not more than 3 to 4Mbps.

The Difference Between Half and Full Duplex Explained

Output-channel of Service Activator

public class SimpleServiceActivatorWithOutput {

@ServiceActivator
public String sayHello(String name) {
return "Hello to " + name;
}
}


<int:channel id="output">
<int:queue/>
</int:channel>

<int:channel id="input"/>

<int:service-activator id="activator" ref="handler" input-channel="input" output-channel="output" />


@Test
public void test() {
input.send(MessageBuilder.withPayload("Jeff").build());
Message<?> r = output.receive();
System.out.println(r.getPayload());
}

If I replace SimpleServiceActivatorWithOutput with SimpleDelayedHandler:


public class SimpleDelayedHandler {

@ServiceActivator
public Object justDelay(Object obj) throws Exception {
int wait = new Random().nextInt(5);
System.out.println("Delays in thread " + Thread.currentThread().getId() + " for " + wait + " secs.");
TimeUnit.SECONDS.sleep(wait);
System.out.println("Ends Delaying in thread " + Thread.currentThread().getId());
return null;
}


I can't get things out of output-channel. But if I change "return null" to "return """, then everything works fine.

Understanding Service Activator

From EIP

• An application has a service that it would like to make available to other application.

Now I understand that this is the context. I am sorry I didn't realize this was part of the Pattern Language (Pattern Language) when I was trying to understand Messaging Gateway.

• How can an application design a service to be invoked both via various messaging technologies and via non-messaging techniques?

This is the System of Forces part. We need to handle messaging technologies and non-messaging technologies. In this way, we need the decoupled solution, right? Talking about decoupling, I would think about Messaging Gateway. However, Messaging Gateway would have a Messaging System on one side while a non-Messaging System on the other. Service Activator doesn't expect this restriction. It just can't tell what would be there on each side, I think.

The problem here is about service invocation. We need to expose the interface to other application. That includes:

• Access point. Like the function name, or alike.
• The parameters.

• Design a Service Activator that connects the messages on the channel to the service being accessed.

Here comes the solution: a guy named Service Activator sitting on the way the messages will come. When the expected message comes by, the Service Activator just invokes the target service.

Does it still supposed to have messaging system on one side? It sounds like an opposite of Messaging Gateway, isn't it?

What about the parameters? What about the return messages if there is any? Since it just wants to invoke a service, doesn't it imply Asynchronous?

• A Service Activator can be one-way (request only) or two-way (Request-Reply). 
• The service can be as simple a method call - synchronous and non-remote - perhaps part of a Service Layer [EAA]. 
• The activator can be hard-coded to always invoke the same service, or can use reflection to invoke the service indicated by the message. 
• The activator handles all of the messaging details and invokes the service like any other client, such that the service doesn't even know it's being invoked through messaging.

This is a very good summary of what characteristics a Service Activator has. From the very last characteristics, it confirms my guess that Service Activator works to handle messaging system request and translates it into a service out of the messaging system.

The Original Service Activator

Core J2EE Patterns

Service Activator was originally named here. 

Problem 

You want to invoke services asynchronously.

Forces

• You want to invoke business services, POJOs, or EJB components in a asynchronous manner.
• You want to integrate publish/subscribe and point-to-point messaging to enable asynchronous processing services.
• You want to perform a business task that is logically composed of several business tasks.

The Half-Sync/Half-Async Pattern in POSA2

• EIP's Service Activator pattern is related to the Half-Sync/Half-Async pattern, which separates service processing into synchronous and asynchronous layers.
• The Half-Sync/Half-Async architectural pattern decouples asynchronous and synchronous service processing in concurrent systems, to simplify programming without unduly reducing performance. The pattern introduces two intercommunicating layers, one for asynchronous and one for synchronous service processing.
• It is hard to develop applications and higher-level system services using asynchrony mechanisms, .... For example, asynchrony can cause subtle timing problems and race conditions when an interrupt preempts a running computation unexpectedly.
• Blocking I/O, in turn, enables developers to maintain state information and execution history implicitly in the run-time stacks of their threads, rather than in separate data structures that must be managed explicitly by developers.
• Within the context of an operating system, however, synchronous and asynchronous processing is not wholly independently.
• A key challenge in the development of ... was the structuring of asynchronous and synchronous processing, to enhance both programming simplicity and system performance.
• Developers of synchronous application programs must be shields from the complex details of asynchronous programming.
• The overall performance of the system must not be degraded by using inefficient synchronous processing mechanisms ...

All descriptions above are very natural. The only thing left to us is the solution.

• Decompose the services in the system into two layers: synchronous and asynchronous, and add a queuing layer between them to mediate the communication between services ...

Really? Before I go further, I would like to recall things that I've done with JBoss Netty. Netty provides a totally event-driven programming model, which is naturally asynchronous. To alleviate the burden of my developers, I encapsulated those Event-Driven API into a set of synchronous API. Did I create a queuing layer in between? No, I didn't. 

I looked into the code again. Oh, yeah, I didn't make everything works synchronously. The major part of the system still relies on Netty's pipeline solution, which is still Event-Driven, or asynchronous. I almost forget about those headache I had.

So, if I had had done those half-sync/half-async things, should I have create a queuing layer in between? And how? I need to think.

• If services residing in separate synchronous and asynchronous layers must communicate or synchronize their processing, allow them to pass messages to each other via a queuing layer.

To be honest, I am lost here. First of all, I think I didn't really design such a system before. For the Netty usage, I did create a data transferring implementation and made it synchronous. What I did was implementing a data sender and a data receiver using Netty's Event-Driven API and then expose these two as a library for my developers. Nothing more. And I don't know why I need more. Yes, this is too simple, right.

I needed to implement a list of commands sending and receiving between the client and server. I didn't make it synchronous because it was not too difficult to just use Netty's API. I built my own layers for the protocol and encapsulated them though. But I still exposed a callback interface for event handler. So it is still Event-Driven.

Oh, wait. They always related synchrony with long-duration, while asynchrony with short-living.

• Services in the synchronous layer run in separate threads or processes that can block while performing operations.
• Services in the asynchronous layer cannot block while performing operations without unduly degrating the performance of other services.

I think this could be a very good example for everyone:

• Many restaurants use a variant of the Half-Sync/Half-Async pattern. For example, restaurants often employ a host or hostess who is responsible for greeting patrons and keeping track of the order in which they will be seated if the restaurant is busy and it is necessary to queue them waiting for an available table. The host or hostess is 'shared' by all the patrons and thus cannot spend much time with any given party. After patrons are seated at a table, a waiter or waitress is dedicated to service that table.

Oh, yeah! That makes a lot of sense. But how Service Activator pattern is related to Half-Sync/Half-Async Architectural Pattern?

How Service Activator Pattern is Related to Half-Sync/Half-Async Architectural Pattern?

Apparently this is from Core J2EE Patterns.

I could understand that Service Activator separates the Synchronous Layer and the Asynchronous Layer, although this is not necessary to be true. But which is the queuing layer? The Messaging System? So that Service Activator encapsulates the service within the hosting application and works as an Asynchronous Layer. It returns the result and put it into the Messaging System for the Client to poll and use in the Synchronous manner? I don't see another way around, because I would believe Asynchronous layer is always the lower layer:

• The asynchronous service layer performs lower-level processing services, which typically emanate from one or more external event sources.

However, in Half-sync/Hals-Async Architecture, the Asynchronous services are supposed to be short-lived, aren't they?

• Services in the asynchronous layer cannot block while performing operations without unduly degrading the performance of the other services.

Service Activator in Spring Integration

• The Service Activator is the endpoint type for connecting any Spring-managed Object to an input channel so that it may play the role of a service.

I really like this description, and I really hate it also.

I like it because it is clear and matches the original definition in EIP very well. We could adapt the context in EIP to something like this here:

• I have a Spring-managed Object and want to use it as a service for another application.

I really hate it, because service is a very ambiguous concept. You can have whatever things as a service. And you can do whatever you want in a service. As long as it could be configured as a Spring-managed Object, which is essentially a Spring IoC bean, and which is basically everything in Spring, you could use Service Activator pattern.

And this is why Service Activator is usually abused. 

I need to admit that Service Activator could be a good one to start with when you can't find a matched candidate. The start point. However, we need to evolve it into something after you find out more. Service Activator, as the name suggest, is the Activator. And as the name implies, it should connect heterogeneous systems, like the Messaging Gateway. If the service it would activate will just add some headers into the message, we should not use Service Activator.

Verifying Multi-threading Feature of Spring Integration Message Endpoints

I wanted to verify that Spring Integration would handle message in a concurrent manner. So I built this sample project.

demo-multithreading.xml


<beans ...>
<bean id="taskExec" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="10" />
</bean>

<int:gateway id="gateway" service-interface="sdemo.SimpleGateway"/>

<bean id="service" class="demo.SimpleDelayedHandler"/>

<int:service-activator input-channel="channel-in" ref="service"/>

</beans>

SimpleGateway.java


public interface SimpleGateway {
@Gateway(requestChannel = "channel-in")
public Message<String> sayHello(Message<String> helloMsg);
}

SimpleDelayedHandler.java

public class SimpleDelayedHandler {

@ServiceActivator
public void justDelay(Object obj) throws Exception {
System.out.println("Delays in thread " + Thread.currentThread().getId());
TimeUnit.SECONDS.sleep(new Random().nextInt(20));
System.out.println("Ends Delaying in thread " + Thread.currentThread().getId());
}
}


Test.java

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath:demo-multithreading.xml" } )
public class Test {

@Autowired SimpleGateway gateway;
@Autowired TaskExecutor taskExec;

@Test
public void test() throws Exception {
Runnable r = new Runnable() {

@Override
public void run() {
System.out.println("Saying hello from thread : " + Thread.currentThread().getId());
gateway.sayHello(MessageBuilder.withPayload("").build());
}
};

for(int k=0; k<10; ++k) {
taskExec.execute(r);
}

TimeUnit.MINUTES.sleep(1);
}
}

Output:


Saying hello from thread : 11
Saying hello from thread : 12
Saying hello from thread : 13
Saying hello from thread : 14
Saying hello from thread : 16
Saying hello from thread : 18
Saying hello from thread : 20
Saying hello from thread : 17
Saying hello from thread : 15
Saying hello from thread : 19
Delays in thread 17
Delays in thread 15
Delays in thread 20
Delays in thread 14
Delays in thread 16
Delays in thread 12
Delays in thread 13
Delays in thread 11
Delays in thread 18
Delays in thread 19
Ends Delaying in thread 12
Ends Delaying in thread 11
Ends Delaying in thread 19
Ends Delaying in thread 14
Ends Delaying in thread 16
Ends Delaying in thread 13
Ends Delaying in thread 20
Ends Delaying in thread 18
Ends Delaying in thread 17
Ends Delaying in thread 15