code: @Override public byte[] apply(byte[] data, int offset, int length) throws Exception {
          ByteArrayDataInputStream in=new ByteArrayDataInputStream(data, offset, length);
          byte command=in.readByte();
          switch(command) {
              case PUT:
                  K key=Util.objectFromStream(in);
                  V val=Util.objectFromStream(in);
                  V old_val=map.put(key, val);
                  notifyPut(key, val, old_val);
                  return old_val == null? null : Util.objectToByteBuffer(old_val);
              case REMOVE:
                  key=Util.objectFromStream(in);
                  old_val=map.remove(key);
                  notifyRemove(key, old_val);
                  return old_val == null? null : Util.objectToByteBuffer(old_val);
              default:
                  throw new IllegalArgumentException("command " + command + " is unknown");
          }
      }
docs: /////////////////////////////////////// StateMachine callbacks /////////////////////////////////////