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
    public void testPrimitiveSerializeRecordToJsonExpandingValue() throws Exception {
        GenericObject genericObject = new GenericObject() {
            @Override
            public SchemaType getSchemaType() {
                return SchemaType.STRING;
            }

            @Override
            public Object getNativeObject() {
                return "message-value";
            }
        };

        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 Schema.STRING;
            }

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

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

            @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\":\"message-value\","
                + "\"properties\":{\"prop-key\":\"prop-value\"},\"eventTime\":1648502845803}");
    }