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 testSerializeRecordToJsonExpandingValue(SchemaType schemaType) throws Exception {
        RecordSchemaBuilder valueSchemaBuilder = org.apache.pulsar.client.api.schema.SchemaBuilder.record("value");
        valueSchemaBuilder.field("c").type(SchemaType.STRING).optional().defaultValue(null);
        valueSchemaBuilder.field("d").type(SchemaType.INT32).optional().defaultValue(null);
        RecordSchemaBuilder udtSchemaBuilder = SchemaBuilder.record("type1");
        udtSchemaBuilder.field("a").type(SchemaType.STRING).optional().defaultValue(null);
        udtSchemaBuilder.field("b").type(SchemaType.BOOLEAN).optional().defaultValue(null);
        udtSchemaBuilder.field("d").type(SchemaType.DOUBLE).optional().defaultValue(null);
        udtSchemaBuilder.field("f").type(SchemaType.FLOAT).optional().defaultValue(null);
        udtSchemaBuilder.field("i").type(SchemaType.INT32).optional().defaultValue(null);
        udtSchemaBuilder.field("l").type(SchemaType.INT64).optional().defaultValue(null);
        GenericSchema<GenericRecord> udtGenericSchema = Schema.generic(udtSchemaBuilder.build(schemaType));
        valueSchemaBuilder.field("e", udtGenericSchema).type(schemaType).optional().defaultValue(null);
        GenericSchema<GenericRecord> valueSchema = Schema.generic(valueSchemaBuilder.build(schemaType));

        GenericRecord valueGenericRecord = valueSchema.newRecordBuilder()
                .set("c", "1")
                .set("d", 1)
                .set("e", udtGenericSchema.newRecordBuilder()
                        .set("a", "a")
                        .set("b", true)
                        .set("d", 1.0)
                        .set("f", 1.0f)
                        .set("i", 1)
                        .set("l", 10L)
                        .build())
                .build();

        Map<String, String> properties = new HashMap<>();
        properties.put("prop-key", "prop-value");

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

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

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

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

            @Override
            public Map<String, String> getProperties() {
                return properties;
            }

            @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\":{\"c\":\"1\","
                + "\"d\":1,\"e\":{\"a\":\"a\",\"b\":true,\"d\":1.0,\"f\":1.0,\"i\":1,\"l\":10}},"
                + "\"properties\":{\"prop-key\":\"prop-value\"},\"eventTime\":1648502845803}");
    }