protected JavaRDD<GenericRecord> maybeAppendKafkaOffsets(JavaRDD<ConsumerRecord<Object, Object>> kafkaRDD) {
    if (this.shouldAddOffsets) {
      AvroConvertor convertor = new AvroConvertor(schemaProvider.getSourceSchema());
      return kafkaRDD.map(convertor::withKafkaFieldsAppended);
    } else {
      return kafkaRDD.map(consumerRecord -> (GenericRecord) consumerRecord.value());
    }
  }
@Test
  public void testAppendKafkaOffsets() throws IOException {
    UtilitiesTestBase.Helpers.saveStringsToDFS(
        new String[] {dataGen.generateGenericRecord().getSchema().toString()}, hoodieStorage(),
        SCHEMA_PATH);
    ConsumerRecord<Object, Object> recordConsumerRecord =
        new ConsumerRecord<Object, Object>("test", 0, 1L,
            "test", dataGen.generateGenericRecord());
    JavaRDD<ConsumerRecord<Object, Object>> rdd =
        jsc().parallelize(Arrays.asList(recordConsumerRecord));
    TypedProperties props = new TypedProperties();
    props.put("hoodie.streamer.source.kafka.topic", "test");
    props.put("hoodie.streamer.schemaprovider.source.schema.file", SCHEMA_PATH);
    SchemaProvider schemaProvider = UtilHelpers.wrapSchemaProviderWithPostProcessor(
        UtilHelpers.createSchemaProvider(FilebasedSchemaProvider.class.getName(), props, jsc()),
        props, jsc(), new ArrayList<>());

    AvroKafkaSource avroKafkaSource =
        new AvroKafkaSource(props, jsc(), spark(), schemaProvider, null);
    GenericRecord withoutKafkaOffsets =
        avroKafkaSource.maybeAppendKafkaOffsets(rdd).collect().get(0);

    props.put(HoodieStreamerConfig.KAFKA_APPEND_OFFSETS.key(), "true");
    schemaProvider = UtilHelpers.wrapSchemaProviderWithPostProcessor(
        UtilHelpers.createSchemaProvider(FilebasedSchemaProvider.class.getName(), props, jsc()),
        props, jsc(), new ArrayList<>());
    avroKafkaSource = new AvroKafkaSource(props, jsc(), spark(), schemaProvider, null);
    GenericRecord withKafkaOffsets = avroKafkaSource.maybeAppendKafkaOffsets(rdd).collect().get(0);
    assertEquals(4, withKafkaOffsets.getSchema().getFields().size()
        - withoutKafkaOffsets.getSchema().getFields().size());
    assertEquals("test", withKafkaOffsets.get("_hoodie_kafka_source_key").toString());

    // scenario with null kafka key
    ConsumerRecord<Object, Object> recordConsumerRecordNullKafkaKey =
        new ConsumerRecord<Object, Object>("test", 0, 1L,
            null, dataGen.generateGenericRecord());
    JavaRDD<ConsumerRecord<Object, Object>> rddNullKafkaKey =
        jsc().parallelize(Arrays.asList(recordConsumerRecordNullKafkaKey));
    avroKafkaSource = new AvroKafkaSource(props, jsc(), spark(), schemaProvider, null);
    GenericRecord withKafkaOffsetsAndNullKafkaKey =
        avroKafkaSource.maybeAppendKafkaOffsets(rddNullKafkaKey).collect().get(0);
    assertNull(withKafkaOffsetsAndNullKafkaKey.get("_hoodie_kafka_source_key"));
  }