public static String serializeRecordToJson(Record<GenericObject> record) {
        checkNotNull(record, "record can't be null");
        final org.apache.pulsar.client.api.Message<GenericObject> recordMessage = getMessage(record);

        JsonObject result = new JsonObject();
        result.addProperty(PAYLOAD_FIELD, getEncoder().encodeToString(recordMessage.getData()));
        if (record.getProperties() != null) {
            JsonObject properties = new JsonObject();
            record.getProperties().forEach(properties::addProperty);
            result.add(PROPERTIES_FIELD, properties);
        }

        final Optional<EncryptionContext> optEncryptionCtx = recordMessage.getEncryptionCtx();
        if (optEncryptionCtx.isPresent()) {
            EncryptionContext encryptionCtx = optEncryptionCtx.get();
            JsonObject encryptionCtxJson = new JsonObject();
            JsonObject keyBase64Map = new JsonObject();
            JsonObject keyMetadataMap = new JsonObject();
            encryptionCtx.getKeys().forEach((key, value) -> {
                keyBase64Map.addProperty(key, getEncoder().encodeToString(value.getKeyValue()));
                Map<String, String> keyMetadata = value.getMetadata();
                if (keyMetadata != null && !keyMetadata.isEmpty()) {
                    JsonObject metadata = new JsonObject();
                    value.getMetadata().forEach(metadata::addProperty);
                    keyMetadataMap.add(key, metadata);
                }
            });
            encryptionCtxJson.add(KEY_MAP_FIELD, keyBase64Map);
            encryptionCtxJson.add(KEY_METADATA_MAP_FIELD, keyMetadataMap);
            encryptionCtxJson.addProperty(ENCRYPTION_PARAM_FIELD,
                    getEncoder().encodeToString(encryptionCtx.getParam()));
            encryptionCtxJson.addProperty(ALGO_FIELD, encryptionCtx.getAlgorithm());
            if (encryptionCtx.getCompressionType() != null) {
                encryptionCtxJson.addProperty(COMPRESSION_TYPE_FIELD, encryptionCtx.getCompressionType().name());
                encryptionCtxJson.addProperty(UNCPRESSED_MSG_SIZE_FIELD, encryptionCtx.getUncompressedMessageSize());
            }
            if (encryptionCtx.getBatchSize().isPresent()) {
                encryptionCtxJson.addProperty(BATCH_SIZE_FIELD, encryptionCtx.getBatchSize().get());
            }
            result.add(ENCRYPTION_CTX_FIELD, encryptionCtxJson);
        }
        return result.toString();
    }
@Test
    public void testJsonSerialization() {

        final String[] keyNames = {"key1", "key2"};
        final String key1Value = "test1";
        final String key2Value = "test2";
        final byte[][] keyValues = {key1Value.getBytes(), key2Value.getBytes()};
        final String param = "param";
        final String algo = "algo";
        int batchSize = 10;
        int compressionMsgSize = 10;

        // serialize to json
        byte[] data = "payload".getBytes();
        Map<String, String> properties = Maps.newHashMap();
        properties.put("prop1", "value");
        Map<String, String> metadata1 = Maps.newHashMap();
        metadata1.put("version", "v1");
        metadata1.put("ckms", "cmks-1");
        Map<String, String> metadata2 = Maps.newHashMap();
        metadata2.put("version", "v2");
        metadata2.put("ckms", "cmks-2");
        Record<GenericObject> recordCtx =
                createRecord(data, algo, keyNames, keyValues, param.getBytes(), metadata1, metadata2,
                        batchSize, compressionMsgSize, properties, true);
        String json = Utils.serializeRecordToJson(recordCtx);

        // deserialize from json and assert
        KinesisMessageResponse kinesisJsonResponse = deSerializeRecordFromJson(json);
        assertEquals(data, getDecoder().decode(kinesisJsonResponse.getPayloadBase64()));
        EncryptionCtx encryptionCtxDeser = kinesisJsonResponse.getEncryptionCtx();
        assertEquals(key1Value.getBytes(),
                getDecoder().decode(encryptionCtxDeser.getKeysMapBase64().get(keyNames[0])));
        assertEquals(key2Value.getBytes(),
                getDecoder().decode(encryptionCtxDeser.getKeysMapBase64().get(keyNames[1])));
        assertEquals(param.getBytes(), getDecoder().decode(encryptionCtxDeser.getEncParamBase64()));
        assertEquals(algo, encryptionCtxDeser.getAlgorithm());
        assertEquals(metadata1, encryptionCtxDeser.getKeysMetadataMap().get(keyNames[0]));
        assertEquals(metadata2, encryptionCtxDeser.getKeysMetadataMap().get(keyNames[1]));
        assertEquals(properties, kinesisJsonResponse.getProperties());

    }