@Override
  public GenericRow extract(Map<String, Object> from, GenericRow to) {
    Set<String> clpEncodedFieldNames = _config.getFieldsForClpEncoding();

    if (_extractAll) {
      for (Map.Entry<String, Object> recordEntry : from.entrySet()) {
        String recordKey = recordEntry.getKey();
        Object recordValue = recordEntry.getValue();
        if (clpEncodedFieldNames.contains(recordKey)) {
          encodeFieldWithClp(recordKey, recordValue, to);
        } else {
          if (null != recordValue) {
            recordValue = convert(recordValue);
          }
          to.putValue(recordKey, recordValue);
        }
      }
      return to;
    }

    // Handle un-encoded fields
    for (String fieldName : _fields) {
      Object value = from.get(fieldName);
      if (null != value) {
        value = convert(value);
      }
      to.putValue(fieldName, value);
    }

    // Handle encoded fields
    for (String fieldName : _config.getFieldsForClpEncoding()) {
      Object value = from.get(fieldName);
      encodeFieldWithClp(fieldName, value, to);
    }
    return to;
  }
@Test
  public void testBadCLPEncodingConfig() {
    Map<String, String> props = new HashMap<>();
    Set<String> fieldsToRead = new HashSet<>();
    // Add some fields for CLP encoding with some mistakenly empty field names
    String separator = FIELDS_FOR_CLP_ENCODING_SEPARATOR;
    props.put(FIELDS_FOR_CLP_ENCODING_CONFIG_KEY, separator + _MESSAGE_1_FIELD_NAME
        + separator + separator + _MESSAGE_2_FIELD_NAME + separator);
    addCLPEncodedField(_MESSAGE_1_FIELD_NAME, fieldsToRead);
    addCLPEncodedField(_MESSAGE_2_FIELD_NAME, fieldsToRead);
    // Add some unencoded fields
    fieldsToRead.add(_TIMESTAMP_FIELD_NAME);

    GenericRow row;

    // Test extracting specific fields
    row = extract(props, fieldsToRead);
    assertEquals(row.getValue(_TIMESTAMP_FIELD_NAME), _TIMESTAMP_FIELD_VALUE);
    assertNull(row.getValue(_LEVEL_FIELD_NAME));
    validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME, _MESSAGE_1_FIELD_VALUE);
    validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME, _MESSAGE_2_FIELD_VALUE);

    // Test extracting all fields
    row = extract(props, null);
    assertEquals(row.getValue(_TIMESTAMP_FIELD_NAME), _TIMESTAMP_FIELD_VALUE);
    assertEquals(row.getValue(_LEVEL_FIELD_NAME), _LEVEL_FIELD_VALUE);
    validateClpEncodedField(row, _MESSAGE_1_FIELD_NAME, _MESSAGE_1_FIELD_VALUE);
    validateClpEncodedField(row, _MESSAGE_2_FIELD_NAME, _MESSAGE_2_FIELD_VALUE);
  }