Send records and retry them if acknowledgement is not received

Question

I am working on a project where I need to consume lot of records and then I am sending these records to some other system which uses zeromq.

Here is the flow:

• Store all the incoming records in a CHM from multiple threads. Records will come at a very high speed.
• From a background thread which runs every 30 seconds, send these records from CHM to zeromq servers.
• After sending each record to zeromq servers, add them to a retry bucket as well so that it can be retried after a particular time if acknowledgment is not received for this record.
• We also have a poller runnable thread which receives acknowledgment from zeromq servers that tells these records have been received so once I get an acknowledgment back, I delete that record from retry bucket so that it doesn't get retried.
• Even if some records are sent multiple times it's ok but it's good to minimize this. I am not sure what is the best way to minimize this in my scenario.

Here is my Processor class in which add method will be called by multiple threads to populate dataHolderByPartitionReference CHM in a thread safe way. And then in the constructor of Processor class, I start the background thread which runs every 30 seconds to push records from same CHM to a zeromq servers by calling the SendToZeroMQ class:

public class Processor {
  private final ScheduledExecutorService executorService = Executors
      .newSingleThreadScheduledExecutor();
  // creating a ListeningExecutorService (Guava) by wrapping a normal ExecutorService (Java)
  private final ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors
      .newCachedThreadPool());        
  private final AtomicReference<ConcurrentHashMap<Integer, ConcurrentLinkedQueue<DataHolder>>> dataHolderByPartitionReference =
      new AtomicReference<>(new ConcurrentHashMap<Integer, ConcurrentLinkedQueue<DataHolder>>());

  private static class Holder {
    private static final Processor INSTANCE = new Processor();
  }

  public static Processor getInstance() {
    return Holder.INSTANCE;
  }

  private Processor() {
    executorService.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        validateAndSendAllPartitions(dataHolderByPartitionReference
            .getAndSet(new ConcurrentHashMap<Integer, ConcurrentLinkedQueue<DataHolder>>()));
      }
    }, 0, 30, TimeUnit.SECONDS);
  }

  // calling validateAndSend in parallel for each partition
  // generally there will be only 5-6 unique partitions max
  private void validateAndSendAllPartitions(
      ConcurrentHashMap<Integer, ConcurrentLinkedQueue<DataHolder>> dataHolderByPartition) {
    List<ListenableFuture<Void>> list = new ArrayList<ListenableFuture<Void>>();
    // For each partition, create an independent thread that will
    // validate the dataHolder and send it to the zeromq servers
    for (Entry<Integer, ConcurrentLinkedQueue<DataHolder>> entry : dataHolderByPartition
        .entrySet()) {
     final int partition = entry.getKey();
      final ConcurrentLinkedQueue<DataHolder> dataHolders = entry.getValue();
      ListenableFuture<Void> future = executor.submit(new Callable<Void>() {
        public Void call() throws Exception {
          validateAndSend(partition, dataHolders);
          return null;
        }
      });
      // Add the future to the list
      list.add(future);
    }
    // We want to know when ALL the threads have completed,
    // so we use a Guava function to turn a list of ListenableFutures
    // into a single ListenableFuture
    ListenableFuture<List<Void>> combinedFutures = Futures.allAsList(list);

    // The get on the combined ListenableFuture will now block until
    // ALL the individual threads have completed work.
    try {
      List<Void> allPartitionDataHolders = combinedFutures.get();
    } catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
      // log error
    } catch (ExecutionException ex) {
      // log error
    }
  }

  private void validateAndSend(final int partition,
      final ConcurrentLinkedQueue<DataHolder> dataHolders) {
    Map<byte[], byte[]> clientKeyBytesAndProcessBytesHolder = new HashMap<>();
    int totalSize = 0;
    while (!dataHolders.isEmpty()) {
      DataHolder dataHolder = dataHolders.poll();
      byte[] clientKeyBytes = dataHolder.getClientKey().getBytes(StandardCharsets.UTF_8);
      if (clientKeyBytes.length > 255)
        continue;
      byte[] processBytes = dataHolder.getProcessBytes();
      int clientKeyLength = clientKeyBytes.length;
      int processBytesLength = processBytes.length;

      int additionalLength = clientKeyLength + processBytesLength;
      if (totalSize + additionalLength > 64000) {
        SendToZeroMQ.getInstance().executeAsync(partition, clientKeyBytesAndProcessBytesHolder);
        clientKeyBytesAndProcessBytesHolder.clear(); // watch out for gc
        totalSize = 0;
      }
      clientKeyBytesAndProcessBytesHolder.put(clientKeyBytes, processBytes);
      totalSize += additionalLength;
    }
    // calling again with remaining values
    SendToZeroMQ.getInstance().executeAsync(partition, clientKeyBytesAndProcessBytesHolder);
  }

  // called by multiple threads to populate dataHolderByPartitionReference CHM
  public void add(final int partition, final DataHolder holder) {
    ConcurrentMap<Integer, ConcurrentLinkedQueue<DataHolder>> dataHolderByPartition =
        dataHolderByPartitionReference.get();
    ConcurrentLinkedQueue<DataHolder> dataHolder =
        dataHolderByPartition.get(partition);
    if (dataHolder == null) {
      dataHolder = Queues.newConcurrentLinkedQueue();
      ConcurrentLinkedQueue<DataHolder> currentDataHolder =
          dataHolderByPartition.putIfAbsent(partition, dataHolder);
      if (currentDataHolder != null)
        dataHolder = currentDataHolder;
    }
    dataHolder.add(holder);
  }
}

Here is my SendToZeroMQ class which sends record to zeromq servers and retry accordingly depending on acknowledgment.

• Firstly it will send record to zeromq servers.
• Then it will add same record to retryBucket which will get retried later on depending on whether acknowledgment is received or not.
• In the same class, I start a background thread which runs every 1 minute to send records again which are in the retry bucket.
• Same class also starts the ResponsePoller thread which will keep running forever to see what records have been acknowledged (which we have sent before) so as soon as records are acknowledged, the ResponsePoller thread will remove those record from retryBucket so that it doesn't get retried.

SendToZeroMQ

public class SendToZeroMQ {
  // do I need these two ScheduledExecutorService or one is sufficient to start my both the thread?
  private final ScheduledExecutorService executorServicePoller = Executors
      .newSingleThreadScheduledExecutor();
  private final ScheduledExecutorService executorService = Executors
      .newSingleThreadScheduledExecutor();
  private final Cache<Long, byte[]> retryBucket = CacheBuilder.newBuilder().maximumSize(10000000)
      .removalListener(RemovalListeners.asynchronous(new CustomListener(), executorService))
      .build();

  private static class Holder {
    private static final SendToZeroMQ INSTANCE = new SendToZeroMQ();
  }

  public static SendToZeroMQ getInstance() {
    return Holder.INSTANCE;
  }

  private SendToZeroMQ() {
    executorServicePoller.submit(new ResponsePoller());
    executorService.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        for (Entry<Long, byte[]> entry : retryBucket.asMap().entrySet()) {
          executeAsync(entry.getKey(), entry.getValue());
        }
      }
    }, 0, 1, TimeUnit.MINUTES);
  }

  public boolean executeAsync(final long address, final byte[] encodedByteArray) {
    Optional<ZMQObj> liveSockets = PoolManager.getInstance().getNextSocket();
    if (!liveSockets.isPresent()) {
      return false;
    }
    return executeAsync(address, encodedByteArray, liveSockets.get().getSocket());
  }

  public boolean executeAsync(final long address, final byte[] encodedByteArray, final Socket socket) {
    ZMsg msg = new ZMsg();
    msg.add(encodedByteArray);
    boolean sent = msg.send(socket);
    msg.destroy();
    // add to retry bucket
    retryBucket.put(address, encodedByteArray);
    return sent;
  }

  public boolean executeAsync(final int partition,
      final Map<byte[], byte[]> clientKeyBytesAndProcessBytesHolder) {
    Optional<ZMQObj> liveSockets = PoolManager.getInstance().getNextSocket();
    if (!liveSockets.isPresent()) {
      return false;
    }         
    Map<Long, byte[]> addressToencodedByteArray = encode(partition, clientKeyBytesAndProcessBytesHolder);
    long address = addressToencodedByteArray.entrySet().iterator().next().getKey();
    byte[] encodedByteArray = addressToencodedByteArray.entrySet().iterator().next().getValue();
    return executeAsync(address, encodedByteArray, liveSockets.get().getSocket());
  }

  private Map<Long, byte[]> encode(final int partition,
      final Map<byte[], byte[]> clientKeyBytesAndProcessBytesHolder) {

    // this address will be unique always
    long address = TestUtils.getAddress();
    Frame frame = new Frame(............);
    byte[] packedByteArray = frame.serialize();
    // this map will always have one entry in it.
    return ImmutableMap.of(address, packedByteArray);
  }

  public void removeFromRetryBucket(final long address) {
    retryBucket.invalidate(address);
  }
}

And here is my ResponsePoller class which waits for the acknowledgment for all those records already sent by the other background thread. If acknowledgement is received, then delete it from the retry bucket so that it doesn't get retried.

public class ResponsePoller implements Runnable {
  private static final Random random = new Random();
  private static final int listenerPort = 8076;

  @Override
  public void run() {
    ZContext ctx = new ZContext();
    Socket client = ctx.createSocket(ZMQ.PULL);

    // Set random identity to make tracing easier
    String identity = String.format("%04X-%04X", random.nextInt(), random.nextInt());
    client.setIdentity(identity.getBytes(ZMQ.CHARSET));
    client.bind("tcp://" + TestUtils.getIPAddress() + ":" + listenerPort);

    PollItem[] items = new PollItem[] {new PollItem(client, Poller.POLLIN)};

    while (!Thread.currentThread().isInterrupted()) {
      // Tick once per second, pulling in arriving messages
      for (int centitick = 0; centitick < 100; centitick++) {
        ZMQ.poll(items, 10);
        if (items[0].isReadable()) {
          ZMsg msg = ZMsg.recvMsg(client);
          Iterator<ZFrame> it = msg.iterator();
          while (it.hasNext()) {
            ZFrame frame = it.next();
            try {
              long address = TestUtils.getAddress(frame.getData());
              // remove from retry bucket since we got the acknowledgment for this record
              SendToZeroMQ.getInstance().removeFromRetryBucket(address);
            } catch (Exception ex) {
              // log error
            } finally {
              frame.destroy();
            }
          }
          msg.destroy();
        }
      }
    }
    ctx.destroy();
  }
}

Is there a better way to design this problem as compared to what I have? I might be breaking Single Responsibility Principle here. I am working with Java 7.

Answer 1

Though I dont understand your whole code, I would suggest pulling the caching responsibility out of your SendToZeroMQ Class.

The first thing, that came into my mind was a modified version of the Memento pattern, while you would not save 'States', but use it as your RetryBucket, as you called it.

This way you could have a clear line between the responsibilities of sending to the zero MQ and the caching of ongoing request, that do not have a response yet.

---

In your SendToZeroMQ class, you have a private final Cache<Long, byte[]> called retryBucket. Pull it out from the "sending class", as it is a "caching algorythm". According to the S.O.L.I.D principle, you should try to make singe responsibility classes.

public enum MyCache { //<--- I like the enum version of singleton pattern better

  INSTANCE;

  private final Cache<MyMessage> retryBucket = CacheBuilder.... //same call

  public void addToBucket (MyMessage toRetry) 
  {
    retryBucket.add(toRetry);
  }

  public void removeFromBucket (MyMessage toRemove) 
  {
    retryBucket.invalidate(toRemove);
  }

  public Set<MyMessage> getRetryBucket ()
  {
    //The cache should not by modified this way!
    return Collections.unmodifiableSet(retryBucket.asSet());
  }
}

The MyMessage class is holding the data:

public class MyMessage  {

  private Long adress;
  private byte[] message;

  //getter and setter are default
}

And the RetryCacheCareTaker is "Singleton enum", whichs only purpose is to resend all entries in the cache:

public enum RetryCacheCareTaker implements Runnable {

  INSTANCE;

  private final ScheduledExecutorService executorService = Executors
    .newSingleThreadScheduledExecutor();

  public void start () 
  {
    executorService.scheduleAtFixedRate (this, 0, 1, TimeUnit.MINUTES);
  }

  @Override
  public void run () 
  {
    Set<MyMessage> messagesToRetry = MyCache.INSTANCE.getRetryBucket();
    messagesToRetry.forEach (e -> SendToZeroMQ.getInstance().executeAsync (e.getAdress, e.getMessage));
  }
}

I have to admit, after trying to apply the Memento pattern, I did not know how to exactly use it in your case as well... Sorry for that.

But with the above example code, you have a single responsible class to hold the data (MyMessage), one single responsible class to cache (MyCache) and a class, whichs only responsibility is to resend the whole retry bucket.

Now you could do something like RetryCacheCareTaker.INSTANCE.start() and the automatic resend algorythm is started. As you refer to the "send method" executeAsync(...), all the "send logic" is in your SendToZeroMQ class, which is nice.

I hope this helps you. I do not know, if there is a Cache.asSet() method, but you should get the idea ;)

Answer 2

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.