This is missing in the code:

Store all the incoming records in a CHM from multiple threads. Records will come at a very high speed.

Who adds work to the buckets? To which buckets?

I might be breaking Single Responsibility Principle here.

You are. But you're unsure. Why?

1. Your naming isn't accurate/representative. Naming gives power and definition. The code should tell you a story.
2. You're misusing several frameworks, derailing them from their original purpose. The ensuing clutter is preventing you from having a clear vision.

Naming & breaking down classes

###Processor What does it process? How? Processor is a fine name for a broad interface with a process() method. You're in a real problem, you have to be more specific.

• How about PartitionPopulator?
  It is suitable because the Processor's constructor is injecting work in the dataHolderByPartitionReference.
• How about DataMessenger ?
  It is suitable because it has a validateAndSend method.
• How about DataValidator ?
  It is suitable because it has a validateAndSend method.
• How about BucketManager?
  It is suitable because it breaks down work into buckets
• Etc.

The problem is, you can't and shouldn't choose. These are distinct responsibilities. Make an Object for each. Correct naming will always lead you to understanding the following problems.

###SendToZeroMQ The name is a Verb, so describes a function. It is an Object, so its name should be a Noun describing what it is.

SendToZeroMQ is sending Byes to ZeroMQ. But it is also starting a Poller, it is also encoding messages, it is also creating Sockets.

executeAsync exists in three flavours. It is fine to have methods have the same name, if they do the same thing with a twist (one takes an optional parameter, the other assumes a default value etc.). However the 3rd one is encoding some stuff which the other don't, so its name should reflect this. The name contains execute which is way too broad: it should tell me itd does stuff with a retryBucket etc. like collectRetries(). The name also contains Async wich is wrong: it is a capability of the caller to execute these in parallel, the method itself has no knowledge of this.

This class should be split in

Framework mis-Usage & clutter

###Useless instance field passing From an Object instance, you're passing its (private, final!) field to one of its own methods. This is unnecessary clutter.

For example the Processor instance has one private final dataHolderByPartitionReference field. In his constructor, it is passing it through the run() method to its own validateAndSendAllPartitions, which of course already had access to that field! This is unnecessary, and it creates usueless additional parameters (which in your case have a very long type definition). In addition, the reader is lead to believe something important is happening, and waste time finding out there is nothing of value.

###Staggered Threading From Processor and using SendToZeroMQ, you are:

• Scheduling Threads to start sequentially at intervals...
• Which must create a bunch of Callables which are called in parallel...
• Each of them must send a piece of work to SendToZeroMQ...
• Which must asynchronously translate those work unit to Byte...
• And put them in series through an executorService to a remote Socket.
• ... In parallel to all of this, ascheduled Poller is poping up and does stuff

That's a bit much. I don't even think half of these processes have a reason to be.

Why pass asynchronously from one class to the other, through parallel buckets of parallel jobs, when you're submitting those job to the Socket in series?

###Future A Future can return a result. You have Future<Void>, so you're not using Futures correctly. You are merely using their ability to be run... So you really need a Runnable.

You could have used their return value for your callback, for example to confirm the message was sent correctly! As it is, the failure cases are handled by Guava handling them as a single Future, so you don't know how each individual job will be affected bo other's failure, and you have no return status, leading you to think you need a result poller.

How to improve

Split the classes around.

I propose you create a Message object, and pass it around to some synchronised queues.

What is the use of the bucket? Usually, you send messages in buckets to reduce the number of messages. But you're sending the messages individually, so this defeats the purpose. Right now the implementation would benefit from not having the buckets.

If you want to use Buckets, then have a Bucket Object but give it the functionality to pass several Messages at once to ZeroMQ somehow. Then have a callback function per Bucket.

Re-think your Threading. You should only need this:

• MessageHandler.receive(Message) Thread-safe method
• MessageHandler.retry(Message) Thread-safe method
• message.sendTo(Socket) synchronous
• ZeroMqMonitor.isMessageReceived() Thread-safe method

Even if Message is actually an entire Bucket, the above should hold.

The code is quite complex and goes back and forth. I hope I've understood things correctly. Take it with a grain of salt.