code: @Override
      public void map(GenericData.Record record,
                      AvroCollector<Pair<ByteBuffer, ByteBuffer>> collector,
                      Reporter reporter) throws IOException {
  
          byte[] keyBytes = null;
          byte[] valBytes = null;
          Object keyRecord = null;
          Object valRecord = null;
          try {
              keyRecord = record.get(keyField);
              valRecord = record.get(valField);
              keyBytes = keySerializer.toBytes(keyRecord);
              valBytes = valueSerializer.toBytes(valRecord);
  
              this.collectorWrapper.setCollector(collector);
              this.mapper.map(keyBytes, valBytes, this.collectorWrapper);
  
              recordCounter++;
          } catch (OutOfMemoryError oom) {
              logger.error(oomErrorMessage(reporter));
              if (keyBytes == null) {
                  logger.error("keyRecord caused OOM!");
              } else {
                  logger.error("keyRecord: " + keyRecord);
                  logger.error("valRecord: " + (valBytes == null ? "caused OOM" : valRecord));
              }
              throw new VoldemortException(oomErrorMessage(reporter), oom);
          }
      }
docs: Create the voldemort key and value from the input Avro record by
  extracting the key and value and map it out for each of the responsible
  voldemort nodes
  
  
  The output value is the node_id & partition_id of the responsible node
  followed by serialized value