Sending data to a database in size-limited chunks

Question

I have a method which takes a parameter which is Partition enum. This method will be called by multiple background threads around same time period by passing different value of partition. Here dataHoldersByPartition is a map of Partition and ConcurrentLinkedQueue<DataHolder>.

  private void validateAndSend(final Partition partition) {  
    ConcurrentLinkedQueue<DataHolder> dataHolders = dataHoldersByPartition.get(partition);
    Map<byte[], byte[]> clientKeyBytesAndProcessBytesHolder = new HashMap<>();
    int totalSize = 0;      
    DataHolder dataHolder;
    while ((dataHolder = dataHolders.poll())  != null) {      
      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 > 50000) {
        Message message = new Message(partition, clientKeyBytesAndProcessBytesHolder);
        sendToDatabase(message.getAddress(), message.getLocation());
        clientKeyBytesAndProcessBytesHolder.clear(); // watch out for gc
        totalSize = 0;
      }
      clientKeyBytesAndProcessBytesHolder.put(clientKeyBytes, processBytes);
      totalSize += additionalLength;
    }
    // calling again with remaining values only if clientKeyBytesAndProcessBytesHolder is not empty
    if(!clientKeyBytesAndProcessBytesHolder.isEmpty()) {
        Message message = new Message(partition, clientKeyBytesAndProcessBytesHolder);
        sendToDatabase(message.getAddress(), message.getLocation());        
    }
  }

In the method, I will iterate dataHolders CLQ and I will extract clientKeyBytes and processBytes from it. Here is the validation that I am supposed to do:

• If the clientKeyBytes length is greater than 255 then I will skip it and continue iterating.
• I will keep incrementing the totalSize variable which will be the sum of clientKeyLength and processBytesLength, and this totalSize length should always be less than 50000 bytes.
• As soon as it reaches the 50000 limit, I will send the clientKeyBytesAndProcessBytesHolder map to the sendToDatabase method and clear out the map, reset totalSize to 0 and start populating again.
• If it doesn't reaches that limit and dataHolders got empty, then it will send whatever it has.

Basically, what I have to make sure is whenever the sendToDatabase method is called, the clientKeyBytesAndProcessBytesHolder map should have size less than 50000 (sum of all keys and values length). It should never be called with the size greater than 50000 (50k).

---

Question:

My code works fine, but as you can see, I am calling the sendToDatabase method at two places: one is as soon as it is reaching the 50k limit, and the other is it didn't reach because we ran out of elements.

Is there any way by which I can call the sendToDatabase method just once and still fulfill both my conditions by rewriting it in a better way? I am working with Java 7.

Answer 1

I waited a few days to get other answers to your questions. now I feel I can add my 2 cents.

---

First off, the obvious: This is one single method. It does:

• Polling a queue
• Serialising a message into a byte[]
• Validating the messages
• Packaging in 50Kb packages
• Sending chunks to a Database

The obvious thing to do is split those actions into individual methods, and move the right methods into the right objects which allow you to use a nicer syntax (right now, too much is in the current single method of the current objec, so your hands are tied).

Our goal is to make your code more readable, like this:

public void batchSend(final Partition partition) {
    Queue<DataHolder> dataHolderQueue = dataHoldersByPartition.get(partition);
    DataHolder data = dataHolderQueue.poll();
    Message message = new Message();
    while (data != null) {
        if(data.isValid()){
            if(message.hasRoomFor(data)){
                message.append(dataHolder);
            } else {
                // Message is at maximum capacity, send it
                database.send(message);
                message = new Message();
            }
        }
        data = dataHolderQueue.poll();
    }
    if(!message.isEmpty()) {
        database.send(message); // Queue is empty, sending remaining message
    }
}

I simply modified the Message Object, which now holds a bunch of DataHolder's serialized data and tells you when you reach a certain size limit. I don't really understand how you concatenate the messages together (it's all hidden in the Message class, which was not included) so most of the below is just a guess:

public class Message {
    public static final int MAX_BYTES_PER_MESSAGE = 50000;
    private final Map<byte[], byte[]> data = new HashMap<>();
    private int size = 0;
    ... Other fields I don't know about ...
    public boolean hasRoomFor(DataHolder holder){
        return size + holder.getSizeIncludingHeader() <= 50000;
    }
    public void append(DataHolder holder){
        byte[] clientKeyBytes = dataHolder.getClientKey().getBytes(StandardCharsets.UTF_8);
        byte[] processBytes = dataHolder.getProcessBytes();
        size += clientKeyBytes.length + processBytes.length;
        // Maybe throw InvalidOperationException is size > 50000 ?
        data.put(clientKeyBytes, processBytes);
    }
    public boolean isEmpty(){
        return map.isEmpty();
    }
    // provide getters on the data field for sendTodatabase() method
    ... Other methods I don't know about ...
}

Also modified the DataHolder Object to add:

public boolean dataHolder.isValid(){
    return dataHolder.getClientKey().getBytes(StandardCharsets.UTF_8) <= 255;
}

Indeed, its validity seem to be an intrinsic property of a DataHolder. By this I mean a DataHolder either is valid, or isn't, but it does not depend on external factors. This is a perfect candidate to be made a method of the DataHolder Object.

---

Weird message usage

This is weird:

   Message message = new Message(partition, clientKeyBytesAndProcessBytesHolder);
   sendToDatabase(message.getAddress(), message.getLocation());

You're creating an instance of message, but not using the instance, only some of its fields, via getters. The point of using a Message Object is to keep its constitutive elements consistent inside, so those elements are only consistent if using the instance.

By allowing sendToDatabase to have two split parameter, you're allowing the caller to pass two inconsistent parameters:

sendToDatabase(anAddress, aTotallyUnrelatedLocation);

The obvious fix is to pass in the Message Object entirely, the method will just extract the right data:

boolean sendToDatabase(Message message){
    Address address = message.getAddress();
    Location location = message.getLocation();
    ... do Stuff
    return successStatus;
}

It is additionally syntaxically more pleasing to send a Message rather than an address and a location.

By the way, why would the message body not be sent as well? This is confusing. I would have expected sendToDatabase(Message message, Address address) at least. Maybe the Location is the Message body? It does not make sense to me.

---

GC Handling

This should never be written:

clientKeyBytesAndProcessBytesHolder.clear(); // watch out for gc

The GC knows his stuff. Don't 'help' him. It knows if the collection needs to be GC'ed, or if the data in it does. This call doesn't even make the memory available!

If the collection must be cleared, it must be for business reasons, not because of GC.

Also be careful, if you clear() it, but the sendToDatabase() spawned a Thread which holds onto the reference a bit longer, you've just deleted all data before it was sent.

The best would be to overwrite the reference to the collection with a new one:

clientKeyBytesAndProcessBytesHolder = new HashMap<>();

With this, you signal to the GC that it can get rid of the Collection (you no longer keep a handle), so when no one else holds a reference either, it will be recycled.

---

Late validation

I wonder why you validate() your messages so late. You only validate them just before sending them. You could do this check earlier, for example when the message is sent to the queues.

This is the "Fail Fast" principle: Don't keep corrupt data in your system, flush them as soon as you can.

---

Edit:

One shortcoming of this code is I might perform several dataholder-to-byte conversions. If those are not complex, it's fine. If they are, here is a Workaround:

Create a ValidatedRawData Object wich contains just the raw byte data.

It would be created by the DataHolder itself in a re-write of validate() method: so:

public class DataHolder {
    ... stuff ...
    /** Now returns null if fails to validate */
    public ValidatedRawData validate(){
        byte[] clientKeyBytes = dataHolder.getClientKey().getBytes(StandardCharsets.UTF_8);
        if (clientKeyBytes.length > 255)
            return null;
        byte[] processBytes = dataHolder.getProcessBytes();
        return new ValidatedRawData(clientKeyBytes, processBytes);
    }
}

ValidatedRawData is just a holder, with Getters.

Answer 2

General

I think you are missing following semantic to achieve what you want:

Are there potential messages to send?

This can be represented by:

potentialMessagesToSend = dataHolder != null || !clientKeyBytesAndProcessBytesHolder.isEmpty()

If you are missing to represent semantics as structure you will have to deal with it with algorithmic as you did.

Refactorings

I omit the "continue" and reformulate (negate) the if-statement as it hindered me to apply the refactorings.

I introduced the "guard"-variable "potentialMessagesToSend" and initialized it to true.

I abstracted from which path a Message was created. If a Message is created it will be sent no matter from where it came.

Code

I do not know if I got it totally right, but you can get the idea:

private void validateAndSend(final Partition partition) {
    
    ConcurrentLinkedQueue<DataHolder> dataHolders = dataHoldersByPartition.get(partition);
    Map<byte[], byte[]> clientKeyBytesAndProcessBytesHolder = new HashMap<>();
    int totalSize = 0;
    boolean potentialMessagesToSend = true;
    
    while (potentialMessagesToSend) {
        
        DataHolder dataHolder = dataHolders.poll();
        potentialMessagesToSend = dataHolder != null || !clientKeyBytesAndProcessBytesHolder.isEmpty();
        
        Message message = null;
        
        if (dataHolder != null) {
            
            byte[] clientKeyBytes = dataHolder.getClientKey().getBytes(StandardCharsets.UTF_8);
            
            if (clientKeyBytes.length <= 255) {
                
                byte[] processBytes = dataHolder.getProcessBytes();
    
                int clientKeyLength = clientKeyBytes.length;
                int processBytesLength = processBytes.length;
    
                int additionalLength = clientKeyLength + processBytesLength;
                
                if (totalSize + additionalLength > 50000) {
                    message = new Message(partition, copy(clientKeyBytesAndProcessBytesHolder));
                    clientKeyBytesAndProcessBytesHolder.clear(); // watch out for gc
                    totalSize = 0;
                }
                
                clientKeyBytesAndProcessBytesHolder.put(clientKeyBytes, processBytes);
                totalSize += additionalLength;
                
            }
            
        } if (!clientKeyBytesAndProcessBytesHolder.isEmpty()) {
            message = new Message(partition, clientKeyBytesAndProcessBytesHolder);
        }
        
        if (message != null) {
            sendToDatabase(message.getAddress(), message.getLocation());
        }
        
    }
    
}

One thing I recognized afterwards. You have to make a copy of the hashmap and then pass it into the Message object. In general it is a good thing to make defensive copies and not pass datastructures through the system by reference.