docs: @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);
          }
      }