code: @Override
      public void write(Record record)
              throws IOException, InterruptedException {
          if (record.isKeyEmpty()) {
              record.setKey(defaultRecord.getKey());
          } else {
              if (record.getKey().isGroupingEmpty() &&
                  !record.getKey().isSortEmpty()) {
                  record.getKey().setGrouping(defaultRecord.getKey().getGrouping());
              }
          }
  
          if (record.isValueEmpty()) {
              record.setValue(defaultRecord.getValue());
          }
  
          flushCount++;
          if (flushCount > cache) {
              flush();
              flushCount = 0;
          }
          List<Value> l = buffer.get(record.getKey());
          if (l == null) {
              l = new ArrayList<Value>();
          }
          l.add(record.getValue());
          buffer.put(record.getKey(), l);
      }
docs: {@inheritDoc}