private String convertToSamzaMessage(SamzaSqlRelRecord relRecord) {
      String jsonValue;
      ObjectNode node = mapper.createObjectNode();

      List<String> fieldNames = relRecord.getFieldNames();
      List<Object> values = relRecord.getFieldValues();

      for (int index = 0; index < fieldNames.size(); index++) {
        Object value = values.get(index);
        if (value == null) {
          continue;
        }

        // TODO limited support right now.
        if (Long.class.isAssignableFrom(value.getClass())) {
          node.put(fieldNames.get(index), (Long) value);
        } else if (Integer.class.isAssignableFrom(value.getClass())) {
          node.put(fieldNames.get(index), (Integer) value);
        } else if (Double.class.isAssignableFrom(value.getClass())) {
          node.put(fieldNames.get(index), (Double) value);
        } else if (String.class.isAssignableFrom(value.getClass())) {
          node.put(fieldNames.get(index), (String) value);
        } else if (SamzaSqlRelRecord.class.isAssignableFrom(value.getClass())) {
          // If the value is a SamzaSqlRelRecord, call convertToSamzaMessage to convert the record to json string.
          node.put(fieldNames.get(index), convertToSamzaMessage((SamzaSqlRelRecord) value));
        } else {
          node.put(fieldNames.get(index), value.toString());
        }
      }
      try {
        jsonValue = mapper.writeValueAsString(node);
      } catch (IOException e) {
        throw new SamzaException("Error json serializing object", e);
      }

      return jsonValue;
    }
SamzaRelConverter that can convert {@link SamzaSqlRelMessage} to json string byte array.
samza-tools/src/main/java/org/apache/samza/tools/json/JsonRelConverterFactory.java