public static String serializeRecordToJsonExpandingValue(ObjectMapper mapper, Record<GenericObject> record,
                                                             boolean flatten)
            throws JsonProcessingException {
        JsonRecord jsonRecord = new JsonRecord();
        GenericObject value = record.getValue();
        if (value != null) {
            jsonRecord.setPayload(toJsonSerializable(record.getSchema(), value.getNativeObject()));
        }
        record.getKey().ifPresent(jsonRecord::setKey);
        record.getTopicName().ifPresent(jsonRecord::setTopicName);
        record.getEventTime().ifPresent(jsonRecord::setEventTime);
        record.getProperties().forEach(jsonRecord::addProperty);

        if (flatten) {
            JsonNode jsonNode = mapper.convertValue(jsonRecord, JsonNode.class);
            return JsonFlattener.flatten(new JacksonJsonValue(jsonNode));
        } else {
            return mapper.writeValueAsString(jsonRecord);
        }
    }
@Test(dataProvider = "schemaType")
    public void testKeyValueSerializeNoValue(SchemaType schemaType) throws Exception {
        RecordSchemaBuilder keySchemaBuilder = org.apache.pulsar.client.api.schema.SchemaBuilder.record("key");
        keySchemaBuilder.field("a").type(SchemaType.STRING).optional().defaultValue(null);
        GenericSchema<GenericRecord> keySchema = Schema.generic(keySchemaBuilder.build(schemaType));

        RecordSchemaBuilder valueSchemaBuilder = org.apache.pulsar.client.api.schema.SchemaBuilder.record("value");
        valueSchemaBuilder.field("c").type(SchemaType.STRING).optional().defaultValue(null);
        GenericSchema<GenericRecord> valueSchema = Schema.generic(valueSchemaBuilder.build(schemaType));

        Schema<org.apache.pulsar.common.schema.KeyValue<GenericRecord, GenericRecord>> keyValueSchema =
                Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE);
        org.apache.pulsar.common.schema.KeyValue<GenericRecord, GenericRecord>
                keyValue = new org.apache.pulsar.common.schema.KeyValue<>(null, null);
        GenericObject genericObject = new GenericObject() {
            @Override
            public SchemaType getSchemaType() {
                return SchemaType.KEY_VALUE;
            }

            @Override
            public Object getNativeObject() {
                return keyValue;
            }
        };

        Record<GenericObject> genericObjectRecord = new Record<>() {
            @Override
            public Optional<String> getTopicName() {
                return Optional.of("data-ks1.table1");
            }

            @Override
            public org.apache.pulsar.client.api.Schema getSchema() {
                return keyValueSchema;
            }

            @Override
            public Optional<String> getKey() {
                return Optional.of("message-key");
            }

            @Override
            public GenericObject getValue() {
                return genericObject;
            }

            @Override
            public Map<String, String> getProperties() {
                return Collections.emptyMap();
            }

            @Override
            public Optional<Long> getEventTime() {
                return Optional.of(1648502845803L);
            }
        };

        ObjectMapper objectMapper = new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL);
        String json = Utils.serializeRecordToJsonExpandingValue(objectMapper, genericObjectRecord, false);

        assertEquals(json, "{\"topicName\":\"data-ks1.table1\",\"key\":\"message-key\","
                + "\"payload\":{},"
                + "\"eventTime\":1648502845803}");

        json = Utils.serializeRecordToJsonExpandingValue(objectMapper, genericObjectRecord, true);

        assertEquals(json, "{\"topicName\":\"data-ks1.table1\",\"key\":\"message-key\","
                + "\"payload\":{},"
                + "\"eventTime\":1648502845803}");
    }