@Nullable
  @Override
  public GenericRow transform(GenericRow record) {
    GenericRow outputRecord = new GenericRow();

    try {
      ExtraFieldsContainer extraFieldsContainer =
          new ExtraFieldsContainer(null != _transformerConfig.getUnindexableExtrasField());
      for (Map.Entry<String, Object> recordEntry : record.getFieldToValueMap().entrySet()) {
        String recordKey = recordEntry.getKey();
        Object recordValue = recordEntry.getValue();
        processField(_schemaTree, recordKey, recordKey, recordValue, extraFieldsContainer, outputRecord);
      }
      putExtrasField(_transformerConfig.getIndexableExtrasField(), _indexableExtrasFieldType,
          extraFieldsContainer.getIndexableExtras(), outputRecord);
      putExtrasField(_transformerConfig.getUnindexableExtrasField(), _unindexableExtrasFieldType,
          extraFieldsContainer.getUnindexableExtras(), outputRecord);
    } catch (Exception e) {
      if (!_continueOnError) {
        throw e;
      }
      _logger.debug("Couldn't transform record: {}", record.toString(), e);
      outputRecord.putValue(GenericRow.INCOMPLETE_RECORD_KEY, true);
    }

    return outputRecord;
  }
@Test
  public void testIgnoringSpecialRowKeys() {
    // Configure a FilterTransformer and a SchemaConformingTransformer such that the filter will introduce a special
    // key $(SKIP_RECORD_KEY$) that the SchemaConformingTransformer should ignore
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setFilterConfig(new FilterConfig("intField = 1"));
    SchemaConformingTransformerConfig schemaConformingTransformerConfig =
        new SchemaConformingTransformerConfig(INDEXABLE_EXTRAS_FIELD_NAME, UNINDEXABLE_EXTRAS_FIELD_NAME,
            UNINDEXABLE_FIELD_SUFFIX, null);
    ingestionConfig.setSchemaConformingTransformerConfig(schemaConformingTransformerConfig);
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").setIngestionConfig(ingestionConfig).build();

    // Create a series of transformers: FilterTransformer -> SchemaConformingTransformer
    List<RecordTransformer> transformers = new LinkedList<>();
    transformers.add(new FilterTransformer(tableConfig));
    Schema schema = createDefaultSchemaBuilder().addSingleValueDimension("intField", DataType.INT).build();
    transformers.add(new SchemaConformingTransformer(tableConfig, schema));
    CompositeTransformer compositeTransformer = new CompositeTransformer(transformers);

    Map<String, Object> inputRecordMap = jsonStringToMap("{\"intField\":1}");
    GenericRow inputRecord = createRowFromMap(inputRecordMap);
    GenericRow outputRecord = compositeTransformer.transform(inputRecord);
    Assert.assertNotNull(outputRecord);
    // Check that the transformed record has $SKIP_RECORD_KEY$
    Assert.assertFalse(IngestionUtils.shouldIngestRow(outputRecord));
  }