code: private Tuple2<Object, T> mapPkToRow(String timestamp, JSONObject jsonRow, List<String> joinFields) {
          Object joinValue = null;
          T rowValues = null;
          try {
              rowValues = rowMapper.newInstance();
              rowValues.getClass().getMethod("setTimestamp", String.class).invoke(rowValues, timestamp);
              for (Object key : jsonRow.keySet()) {
                  String colValue = key.toString();
                  Util.applyKVToBean(rowValues, colValue, jsonRow.get(colValue));
                  if (joinFields.contains(colValue)) {
                      joinValue = (joinValue == null)?jsonRow.get(colValue):(joinValue + "\u0001" + jsonRow.get(colValue));
                  }
              }
              if (joinFields.contains("timestamp")) {// Join field could contain timestamp(timestamp can be out of the actual data JSON, so we try this way)
                  joinValue = (joinValue == null)?timestamp:(joinValue + "\u0001" + timestamp);
              }
          } catch (InstantiationException | IllegalAccessException | NoSuchMethodException | SecurityException | IllegalArgumentException | InvocationTargetException ex) {
              Logger.getLogger(Joiner4Bean.class.getName()).log(Level.SEVERE, null, ex);
          }
          return new Tuple2<>(joinValue, rowValues);
      }
docs: Extract fields(k,v) from json
  k = primary field(s), i.e could be composite key as well.
  v = all fields . The first field is always timestamp.
  @param timestamp
  @param jsonRow
  @param joinFields
  @return