OpenCsvRecordMarshaller implements RecordMarshaller<P, String> { @Override public StringRecord processRecord(Record<P> record) throws Exception { try (StringWriter stringWriter = new StringWriter(); CSVWriter csvWriter = new CSVWriter(stringWriter, delimiter, qualifier, "")) { P payload = record.getPayload(); List<String> fields = extractFields(payload); String[] items = fields.toArray(new String[0]); csvWriter.writeNext(items); csvWriter.flush(); return new StringRecord(record.getHeader(), stringWriter.toString()); } } OpenCsvRecordMarshaller(final Class<P> type, final String... fields);  OpenCsvRecordMarshaller(final FieldExtractor<P> fieldExtractor); @Override StringRecord processRecord(Record<P> record); void setDelimiter(char delimiter); void setQualifier(char qualifier); static final char DEFAULT_DELIMITER; static final char DEFAULT_QUALIFIER; }
@Test public void processRecord() throws Exception { String expectedPayload = "\"foo\",\"bar\",\"30\",\"true\""; StringRecord actual = marshaller.processRecord(record); assertThat(actual.getHeader()).isEqualTo(header); assertThat(actual.getPayload()).isEqualTo(expectedPayload); }