pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rdhaba...@apache.org
Subject [incubator-pulsar] branch master updated: add flatbuffer option to serialize kinesis-message in KinesisSink (#2108)
Date Thu, 12 Jul 2018 23:59:04 GMT
This is an automated email from the ASF dual-hosted git repository.

rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 2a6d195  add flatbuffer option to serialize kinesis-message in KinesisSink (#2108)
2a6d195 is described below

commit 2a6d1950954e9d0b1ec33c7fc738ebbdab0a4b9e
Author: Rajan Dhabalia <rdhabalia@apache.org>
AuthorDate: Thu Jul 12 16:59:00 2018 -0700

    add flatbuffer option to serialize kinesis-message in KinesisSink (#2108)
    
    * add flatbuffer option to serialize kinesis-message
    
    * add flatbuffer license
    
    * fix license
---
 distribution/server/src/assemble/LICENSE.bin.txt   |   2 +
 pom.xml                                            |  12 ++
 .../pulsar/common/api/EncryptionContext.java       |   1 -
 pulsar-io/kinesis/pom.xml                          |   6 +
 .../kinesis/src/main/fb/KinesisMessageApi.fbs      |  59 +++++++
 .../org/apache/pulsar/io/kinesis/KinesisSink.java  |  10 +-
 .../pulsar/io/kinesis/KinesisSinkConfig.java       |   6 +-
 .../java/org/apache/pulsar/io/kinesis/Utils.java   | 129 ++++++++++++++--
 .../pulsar/io/kinesis/fbs/CompressionType.java     |  15 ++
 .../pulsar/io/kinesis/fbs/EncryptionCtx.java       |  68 +++++++++
 .../pulsar/io/kinesis/fbs/EncryptionKey.java       |  52 +++++++
 .../org/apache/pulsar/io/kinesis/fbs/KeyValue.java |  41 +++++
 .../org/apache/pulsar/io/kinesis/fbs/Message.java  |  53 +++++++
 .../org/apache/pulsar/io/kinesis/UtilsTest.java    | 170 ++++++++++++++++-----
 14 files changed, 572 insertions(+), 52 deletions(-)

diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index 763f62c..33b9748 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -459,6 +459,8 @@ The Apache Software License, Version 2.0
     - org.inferred-freebuilder-1.14.9.jar
   * Snappy Java
     - org.xerial.snappy-snappy-java-1.1.1.3.jar
+  * Flatbuffers Java
+    - com.google.flatbuffers-flatbuffers-java-1.9.0.jar
 
 
 BSD 3-clause "New" or "Revised" License
diff --git a/pom.xml b/pom.xml
index ba705a7..8a0da98 100644
--- a/pom.xml
+++ b/pom.xml
@@ -946,6 +946,11 @@ flexible messaging model and an intuitive client API.</description>
             <exclude>src/main/java/org/apache/bookkeeper/mledger/proto/MLDataFormats.java</exclude>
             <exclude>src/main/java/org/apache/pulsar/broker/service/schema/proto/SchemaRegistryFormat.java</exclude>
             <exclude>src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/CompressionType.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionCtx.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionKey.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/KeyValue.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/Message.java</exclude>
             <exclude>**/ByteBufCodedInputStream.java</exclude>
             <exclude>**/ByteBufCodedOutputStream.java</exclude>
             <exclude>bin/proto/*</exclude>
@@ -1050,6 +1055,13 @@ flexible messaging model and an intuitive client API.</description>
             <exclude>src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java</exclude>
             <exclude>src/main/java/org/apache/pulsar/broker/service/schema/proto/SchemaRegistryFormat.java</exclude>
             <exclude>bin/proto/MLDataFormats_pb2.py</exclude>
+            
+            <!-- pulasr-io-connector kinesis : auto generated files from flatbuffer schema
-->
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/CompressionType.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionCtx.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionKey.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/KeyValue.java</exclude>
+            <exclude>src/main/java/org/apache/pulsar/io/kinesis/fbs/Message.java</exclude>
 
             <!-- These files are BSD licensed files -->
             <exclude>src/main/java/org/apache/pulsar/common/util/protobuf/ByteBufCodedInputStream.java</exclude>
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/EncryptionContext.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/api/EncryptionContext.java
index 98eaad7..ff359c5 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/EncryptionContext.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/EncryptionContext.java
@@ -34,7 +34,6 @@ public class EncryptionContext {
 
     private Map<String, EncryptionKey> keys;
     private byte[] param;
-    private Map<String, String> metadata;
     private String algorithm;
     private CompressionType compressionType;
     private int uncompressedMessageSize;
diff --git a/pulsar-io/kinesis/pom.xml b/pulsar-io/kinesis/pom.xml
index 08c3004..f0c2776 100644
--- a/pulsar-io/kinesis/pom.xml
+++ b/pulsar-io/kinesis/pom.xml
@@ -81,6 +81,12 @@
       <version>0.12.8</version>
     </dependency>
 	<!-- /kinesis dependencies -->
+	
+    <dependency>
+      <groupId>com.google.flatbuffers</groupId>
+      <artifactId>flatbuffers-java</artifactId>
+      <version>1.9.0</version>
+    </dependency>
 
   </dependencies>
 
diff --git a/pulsar-io/kinesis/src/main/fb/KinesisMessageApi.fbs b/pulsar-io/kinesis/src/main/fb/KinesisMessageApi.fbs
new file mode 100644
index 0000000..f7cc030
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/fb/KinesisMessageApi.fbs
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+ 
+ /**
+ * Instrunction to generate fb-source
+ * export PULSAR_HOME=<Path where you cloned the pulsar repo>
+ * export KINESIS_IO_MAIN=${PULSAR_HOME}/pulsar-io/kinesis/src/main
+ * Command to build java src:  flatc --java  -o ${KINESIS_IO_MAIN}/java ${KINESIS_IO_MAIN}/fb/KinesisMessageApi.fbs

+ * flatc version 1.9.0 (pip install flatbuffers)
+ */
+
+namespace org.apache.pulsar.io.kinesis.fbs;
+
+table KeyValue {
+  key:string;
+  value:string;
+}
+
+table EncryptionKey {
+  key:string;
+  value:[byte];
+  metadata:[KeyValue];
+}
+
+enum CompressionType : byte { NONE, LZ4, ZLIB }
+
+table EncryptionCtx {
+  keys:[EncryptionKey];
+  param:[byte];
+  algo:string;
+  compressionType:CompressionType;
+  uncompressedMessageSize:int;
+  batchSize:int;
+  isBatchMessage:bool=false;
+}
+
+table Message {
+  encryptionCtx:EncryptionCtx;
+  properties:[KeyValue];
+  payload:[byte];
+}
+
+root_type Message;
\ No newline at end of file
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
index 5387a66..dbe1e75 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
@@ -103,7 +103,7 @@ public class KinesisSink implements Sink<byte[]> {
                 : partitionedKey; // partitionedKey Length must be at least one, and at most
256
         ListenableFuture<UserRecordResult> addRecordResult = kinesisProducer.addUserRecord(this.streamName,
                 partitionedKey,
-                ByteBuffer.wrap(createKinesisMessage(kinesisSinkConfig.getMessageFormat(),
inputRecordContext, value)));
+                createKinesisMessage(kinesisSinkConfig.getMessageFormat(), inputRecordContext,
value));
         addCallback(addRecordResult,
                 ProducerSendCallback.create(this.streamName, inputRecordContext, System.nanoTime()),
directExecutor());
         if (LOG.isDebugEnabled()) {
@@ -274,12 +274,14 @@ public class KinesisSink implements Sink<byte[]> {
         };
     }
 
-    public static byte[] createKinesisMessage(MessageFormat msgFormat, RecordContext recordCtx,
byte[] data) {
+    public static ByteBuffer createKinesisMessage(MessageFormat msgFormat, RecordContext
recordCtx, byte[] data) {
         if (MessageFormat.FULL_MESSAGE_IN_JSON.equals(msgFormat)) {
-            return Utils.serializeRecordToJson(recordCtx, data).getBytes();
+            return ByteBuffer.wrap(Utils.serializeRecordToJson(recordCtx, data).getBytes());
+        } else if (MessageFormat.FULL_MESSAGE_IN_FB.equals(msgFormat)) {
+            return Utils.serializeRecordToFlatBuffer(recordCtx, data);
         } else {
             // send raw-message
-            return data;
+            return ByteBuffer.wrap(data);
         }
     }
     
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
index e3c8fde..bf5f2ea 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
@@ -76,6 +76,10 @@ public class KinesisSinkConfig implements Serializable {
          * 
          * 
          */
-        FULL_MESSAGE_IN_JSON;
+        FULL_MESSAGE_IN_JSON,
+        /**
+         * Kinesis sink sends message serialized in flat-buffer.
+         */
+        FULL_MESSAGE_IN_FB;
     }
 }
\ No newline at end of file
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/Utils.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/Utils.java
index 8236080..469151e 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/Utils.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/Utils.java
@@ -19,13 +19,22 @@
 
 package org.apache.pulsar.io.kinesis;
 
+import static com.google.common.base.Preconditions.checkNotNull;
 import static java.util.Base64.getEncoder;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
 
 import org.apache.pulsar.common.api.EncryptionContext;
 import org.apache.pulsar.io.core.RecordContext;
+import org.apache.pulsar.io.kinesis.fbs.EncryptionCtx;
+import org.apache.pulsar.io.kinesis.fbs.EncryptionKey;
+import org.apache.pulsar.io.kinesis.fbs.KeyValue;
+import org.apache.pulsar.io.kinesis.fbs.Message;
 
+import com.google.flatbuffers.FlatBufferBuilder;
 import com.google.gson.JsonObject;
 
 public class Utils {
@@ -34,13 +43,121 @@ public class Utils {
     private static final String PROPERTIES_FIELD = "properties";
     private static final String KEY_MAP_FIELD = "keysMapBase64";
     private static final String KEY_METADATA_MAP_FIELD = "keysMetadataMap";
-    private static final String METADATA_FIELD = "metadata";
     private static final String ENCRYPTION_PARAM_FIELD = "encParamBase64";
     private static final String ALGO_FIELD = "algorithm";
     private static final String COMPRESSION_TYPE_FIELD = "compressionType";
     private static final String UNCPRESSED_MSG_SIZE_FIELD = "uncompressedMessageSize";
     private static final String BATCH_SIZE_FIELD = "batchSize";
     private static final String ENCRYPTION_CTX_FIELD = "encryptionCtx";
+    
+    private static final FlatBufferBuilder DEFAULT_FB_BUILDER = new FlatBufferBuilder(0);
+
+    /**
+     * Serialize record to flat-buffer. it's not a thread-safe method.
+     * 
+     * @param inputRecordContext
+     * @param data
+     * @return
+     */
+    public static ByteBuffer serializeRecordToFlatBuffer(RecordContext inputRecordContext,
byte[] data) {
+        DEFAULT_FB_BUILDER.clear();
+        return serializeRecordToFlatBuffer(DEFAULT_FB_BUILDER, inputRecordContext, data);
+    }
+    
+    public static ByteBuffer serializeRecordToFlatBuffer(FlatBufferBuilder builder, RecordContext
inputRecordContext, byte[] data) {
+        checkNotNull(inputRecordContext, "record-context can't be null");
+        Optional<EncryptionContext> encryptionCtx = inputRecordContext.getEncryptionCtx();
+        Map<String, String> properties = inputRecordContext.getProperties();
+
+        int encryptionCtxOffset = -1;
+        int propertiesOffset = -1;
+
+        if (properties != null && !properties.isEmpty()) {
+            int[] propertiesOffsetArray = new int[properties.size()];
+            int i = 0;
+            for (Entry<String, String> property : properties.entrySet()) {
+                propertiesOffsetArray[i++] = KeyValue.createKeyValue(builder, builder.createString(property.getKey()),
+                        builder.createString(property.getValue()));
+            }
+            propertiesOffset = Message.createPropertiesVector(builder, propertiesOffsetArray);
+        }
+
+        if (encryptionCtx.isPresent()) {
+            encryptionCtxOffset = createEncryptionCtxOffset(builder, encryptionCtx);
+        }
+
+        int payloadOffset = Message.createPayloadVector(builder, data);
+        Message.startMessage(builder);
+        Message.addPayload(builder, payloadOffset);
+        if (encryptionCtxOffset != -1) {
+            Message.addEncryptionCtx(builder, encryptionCtxOffset);
+        }
+        if (propertiesOffset != -1) {
+            Message.addProperties(builder, propertiesOffset);
+        }
+        int endMessage = Message.endMessage(builder);
+        builder.finish(endMessage);
+        ByteBuffer bb = builder.dataBuffer();
+        
+        // to avoid copying of data, use same byte[] wrapped by ByteBuffer. But, ByteBuffer.array()
returns entire array
+        // so, it requires to read from offset:
+        // builder.sizedByteArray()=>copies buffer: sizedByteArray(space, bb.capacity()
- space)
+        int space = bb.capacity() - builder.offset(); 
+        return ByteBuffer.wrap(bb.array(), space, bb.capacity() - space);
+    }
+
+    private static int createEncryptionCtxOffset(final FlatBufferBuilder builder, Optional<EncryptionContext>
encryptionCtx) {
+        if (!encryptionCtx.isPresent()) {
+            return -1;
+        }
+        // Message.addEncryptionCtx(builder, encryptionCtxOffset);
+        EncryptionContext ctx = encryptionCtx.get();
+        int[] keysOffsets = new int[ctx.getKeys().size()];
+        int keyIndex = 0;
+        for (Entry<String, org.apache.pulsar.common.api.EncryptionContext.EncryptionKey>
entry : ctx.getKeys()
+                .entrySet()) {
+            int key = builder.createString(entry.getKey());
+            int value = EncryptionKey.createValueVector(builder, entry.getValue().getKeyValue());
+            Map<String, String> metadata = entry.getValue().getMetadata();
+            int[] metadataOffsets = new int[metadata.size()];
+            int i = 0;
+            for (Entry<String, String> m : metadata.entrySet()) {
+                metadataOffsets[i++] = KeyValue.createKeyValue(builder, builder.createString(m.getKey()),
+                        builder.createString(m.getValue()));
+            }
+            int metadataOffset = -1;
+            if (metadata.size() > 0) {
+                metadataOffset = EncryptionKey.createMetadataVector(builder, metadataOffsets);
+            }
+            EncryptionKey.startEncryptionKey(builder);
+            EncryptionKey.addKey(builder, key);
+            EncryptionKey.addValue(builder, value);
+            if(metadataOffset!=-1) {
+                EncryptionKey.addMetadata(builder, metadataOffset);                
+            }
+            keysOffsets[keyIndex++] = EncryptionKey.endEncryptionKey(builder);
+        }
+
+        int keysOffset = EncryptionCtx.createKeysVector(builder, keysOffsets);
+        int param = EncryptionCtx.createParamVector(builder, ctx.getParam());
+        int algo = builder.createString(ctx.getAlgorithm());
+        int batchSize = ctx.getBatchSize().isPresent() ? ctx.getBatchSize().get() : 1;
+        byte compressionType;
+        switch (ctx.getCompressionType()) {
+        case LZ4:
+            compressionType = org.apache.pulsar.io.kinesis.fbs.CompressionType.LZ4;
+            break;
+        case ZLIB:
+            compressionType = org.apache.pulsar.io.kinesis.fbs.CompressionType.ZLIB;
+            break;
+        default:
+            compressionType = org.apache.pulsar.io.kinesis.fbs.CompressionType.NONE;
+
+        }
+        return EncryptionCtx.createEncryptionCtx(builder, keysOffset, param, algo, compressionType,
+                ctx.getUncompressedMessageSize(), batchSize, ctx.getBatchSize().isPresent());
+    
+    }
 
     /**
      * Serializes sink-record into json format. It encodes encryption-keys, encryption-param
and payload in base64
@@ -51,9 +168,7 @@ public class Utils {
      * @return
      */
     public static String serializeRecordToJson(RecordContext inputRecordContext, byte[] data)
{
-        if (inputRecordContext == null) {
-            return null;
-        }
+        checkNotNull(inputRecordContext, "record-context can't be null");
         JsonObject result = new JsonObject();
         result.addProperty(PAYLOAD_FIELD, getEncoder().encodeToString(data));
         if (inputRecordContext.getProperties() != null) {
@@ -79,12 +194,6 @@ public class Utils {
             });
             encryptionCtxJson.add(KEY_MAP_FIELD, keyBase64Map);
             encryptionCtxJson.add(KEY_METADATA_MAP_FIELD, keyMetadataMap);
-            Map<String, String> metadataMap = encryptionCtx.getMetadata();
-            if (metadataMap != null && !metadataMap.isEmpty()) {
-                JsonObject metadata = new JsonObject();
-                encryptionCtx.getMetadata().entrySet().forEach(m -> metadata.addProperty(m.getKey(),
m.getValue()));
-                encryptionCtxJson.add(METADATA_FIELD, metadata);
-            }
             encryptionCtxJson.addProperty(ENCRYPTION_PARAM_FIELD,
                     getEncoder().encodeToString(encryptionCtx.getParam()));
             encryptionCtxJson.addProperty(ALGO_FIELD, encryptionCtx.getAlgorithm());
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/CompressionType.java
b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/CompressionType.java
new file mode 100644
index 0000000..0a90a22
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/CompressionType.java
@@ -0,0 +1,15 @@
+// automatically generated by the FlatBuffers compiler, do not modify
+
+package org.apache.pulsar.io.kinesis.fbs;
+
+public final class CompressionType {
+  private CompressionType() { }
+  public static final byte NONE = 0;
+  public static final byte LZ4 = 1;
+  public static final byte ZLIB = 2;
+
+  public static final String[] names = { "NONE", "LZ4", "ZLIB", };
+
+  public static String name(int e) { return names[e]; }
+}
+
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionCtx.java
b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionCtx.java
new file mode 100644
index 0000000..e6dff72
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionCtx.java
@@ -0,0 +1,68 @@
+// automatically generated by the FlatBuffers compiler, do not modify
+
+package org.apache.pulsar.io.kinesis.fbs;
+
+import java.nio.*;
+import java.lang.*;
+import java.util.*;
+import com.google.flatbuffers.*;
+
+@SuppressWarnings("unused")
+public final class EncryptionCtx extends Table {
+  public static EncryptionCtx getRootAsEncryptionCtx(ByteBuffer _bb) { return getRootAsEncryptionCtx(_bb,
new EncryptionCtx()); }
+  public static EncryptionCtx getRootAsEncryptionCtx(ByteBuffer _bb, EncryptionCtx obj) {
_bb.order(ByteOrder.LITTLE_ENDIAN); return (obj.__assign(_bb.getInt(_bb.position()) + _bb.position(),
_bb)); }
+  public void __init(int _i, ByteBuffer _bb) { bb_pos = _i; bb = _bb; }
+  public EncryptionCtx __assign(int _i, ByteBuffer _bb) { __init(_i, _bb); return this; }
+
+  public EncryptionKey keys(int j) { return keys(new EncryptionKey(), j); }
+  public EncryptionKey keys(EncryptionKey obj, int j) { int o = __offset(4); return o !=
0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
+  public int keysLength() { int o = __offset(4); return o != 0 ? __vector_len(o) : 0; }
+  public byte param(int j) { int o = __offset(6); return o != 0 ? bb.get(__vector(o) + j
* 1) : 0; }
+  public int paramLength() { int o = __offset(6); return o != 0 ? __vector_len(o) : 0; }
+  public ByteBuffer paramAsByteBuffer() { return __vector_as_bytebuffer(6, 1); }
+  public ByteBuffer paramInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb,
6, 1); }
+  public String algo() { int o = __offset(8); return o != 0 ? __string(o + bb_pos) : null;
}
+  public ByteBuffer algoAsByteBuffer() { return __vector_as_bytebuffer(8, 1); }
+  public ByteBuffer algoInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb,
8, 1); }
+  public byte compressionType() { int o = __offset(10); return o != 0 ? bb.get(o + bb_pos)
: 0; }
+  public int uncompressedMessageSize() { int o = __offset(12); return o != 0 ? bb.getInt(o
+ bb_pos) : 0; }
+  public int batchSize() { int o = __offset(14); return o != 0 ? bb.getInt(o + bb_pos) :
0; }
+  public boolean isBatchMessage() { int o = __offset(16); return o != 0 ? 0!=bb.get(o + bb_pos)
: false; }
+
+  public static int createEncryptionCtx(FlatBufferBuilder builder,
+      int keysOffset,
+      int paramOffset,
+      int algoOffset,
+      byte compressionType,
+      int uncompressedMessageSize,
+      int batchSize,
+      boolean isBatchMessage) {
+    builder.startObject(7);
+    EncryptionCtx.addBatchSize(builder, batchSize);
+    EncryptionCtx.addUncompressedMessageSize(builder, uncompressedMessageSize);
+    EncryptionCtx.addAlgo(builder, algoOffset);
+    EncryptionCtx.addParam(builder, paramOffset);
+    EncryptionCtx.addKeys(builder, keysOffset);
+    EncryptionCtx.addIsBatchMessage(builder, isBatchMessage);
+    EncryptionCtx.addCompressionType(builder, compressionType);
+    return EncryptionCtx.endEncryptionCtx(builder);
+  }
+
+  public static void startEncryptionCtx(FlatBufferBuilder builder) { builder.startObject(7);
}
+  public static void addKeys(FlatBufferBuilder builder, int keysOffset) { builder.addOffset(0,
keysOffset, 0); }
+  public static int createKeysVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4,
data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]);
return builder.endVector(); }
+  public static void startKeysVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4,
numElems, 4); }
+  public static void addParam(FlatBufferBuilder builder, int paramOffset) { builder.addOffset(1,
paramOffset, 0); }
+  public static int createParamVector(FlatBufferBuilder builder, byte[] data) { builder.startVector(1,
data.length, 1); for (int i = data.length - 1; i >= 0; i--) builder.addByte(data[i]); return
builder.endVector(); }
+  public static void startParamVector(FlatBufferBuilder builder, int numElems) { builder.startVector(1,
numElems, 1); }
+  public static void addAlgo(FlatBufferBuilder builder, int algoOffset) { builder.addOffset(2,
algoOffset, 0); }
+  public static void addCompressionType(FlatBufferBuilder builder, byte compressionType)
{ builder.addByte(3, compressionType, 0); }
+  public static void addUncompressedMessageSize(FlatBufferBuilder builder, int uncompressedMessageSize)
{ builder.addInt(4, uncompressedMessageSize, 0); }
+  public static void addBatchSize(FlatBufferBuilder builder, int batchSize) { builder.addInt(5,
batchSize, 0); }
+  public static void addIsBatchMessage(FlatBufferBuilder builder, boolean isBatchMessage)
{ builder.addBoolean(6, isBatchMessage, false); }
+  public static int endEncryptionCtx(FlatBufferBuilder builder) {
+    int o = builder.endObject();
+    return o;
+  }
+}
+
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionKey.java
b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionKey.java
new file mode 100644
index 0000000..44d74f4
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/EncryptionKey.java
@@ -0,0 +1,52 @@
+// automatically generated by the FlatBuffers compiler, do not modify
+
+package org.apache.pulsar.io.kinesis.fbs;
+
+import java.nio.*;
+import java.lang.*;
+import java.util.*;
+import com.google.flatbuffers.*;
+
+@SuppressWarnings("unused")
+public final class EncryptionKey extends Table {
+  public static EncryptionKey getRootAsEncryptionKey(ByteBuffer _bb) { return getRootAsEncryptionKey(_bb,
new EncryptionKey()); }
+  public static EncryptionKey getRootAsEncryptionKey(ByteBuffer _bb, EncryptionKey obj) {
_bb.order(ByteOrder.LITTLE_ENDIAN); return (obj.__assign(_bb.getInt(_bb.position()) + _bb.position(),
_bb)); }
+  public void __init(int _i, ByteBuffer _bb) { bb_pos = _i; bb = _bb; }
+  public EncryptionKey __assign(int _i, ByteBuffer _bb) { __init(_i, _bb); return this; }
+
+  public String key() { int o = __offset(4); return o != 0 ? __string(o + bb_pos) : null;
}
+  public ByteBuffer keyAsByteBuffer() { return __vector_as_bytebuffer(4, 1); }
+  public ByteBuffer keyInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb,
4, 1); }
+  public byte value(int j) { int o = __offset(6); return o != 0 ? bb.get(__vector(o) + j
* 1) : 0; }
+  public int valueLength() { int o = __offset(6); return o != 0 ? __vector_len(o) : 0; }
+  public ByteBuffer valueAsByteBuffer() { return __vector_as_bytebuffer(6, 1); }
+  public ByteBuffer valueInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb,
6, 1); }
+  public KeyValue metadata(int j) { return metadata(new KeyValue(), j); }
+  public KeyValue metadata(KeyValue obj, int j) { int o = __offset(8); return o != 0 ? obj.__assign(__indirect(__vector(o)
+ j * 4), bb) : null; }
+  public int metadataLength() { int o = __offset(8); return o != 0 ? __vector_len(o) : 0;
}
+
+  public static int createEncryptionKey(FlatBufferBuilder builder,
+      int keyOffset,
+      int valueOffset,
+      int metadataOffset) {
+    builder.startObject(3);
+    EncryptionKey.addMetadata(builder, metadataOffset);
+    EncryptionKey.addValue(builder, valueOffset);
+    EncryptionKey.addKey(builder, keyOffset);
+    return EncryptionKey.endEncryptionKey(builder);
+  }
+
+  public static void startEncryptionKey(FlatBufferBuilder builder) { builder.startObject(3);
}
+  public static void addKey(FlatBufferBuilder builder, int keyOffset) { builder.addOffset(0,
keyOffset, 0); }
+  public static void addValue(FlatBufferBuilder builder, int valueOffset) { builder.addOffset(1,
valueOffset, 0); }
+  public static int createValueVector(FlatBufferBuilder builder, byte[] data) { builder.startVector(1,
data.length, 1); for (int i = data.length - 1; i >= 0; i--) builder.addByte(data[i]); return
builder.endVector(); }
+  public static void startValueVector(FlatBufferBuilder builder, int numElems) { builder.startVector(1,
numElems, 1); }
+  public static void addMetadata(FlatBufferBuilder builder, int metadataOffset) { builder.addOffset(2,
metadataOffset, 0); }
+  public static int createMetadataVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4,
data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]);
return builder.endVector(); }
+  public static void startMetadataVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4,
numElems, 4); }
+  public static int endEncryptionKey(FlatBufferBuilder builder) {
+    int o = builder.endObject();
+    return o;
+  }
+}
+
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/KeyValue.java
b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/KeyValue.java
new file mode 100644
index 0000000..8cb53b4
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/KeyValue.java
@@ -0,0 +1,41 @@
+// automatically generated by the FlatBuffers compiler, do not modify
+
+package org.apache.pulsar.io.kinesis.fbs;
+
+import java.nio.*;
+import java.lang.*;
+import java.util.*;
+import com.google.flatbuffers.*;
+
+@SuppressWarnings("unused")
+public final class KeyValue extends Table {
+  public static KeyValue getRootAsKeyValue(ByteBuffer _bb) { return getRootAsKeyValue(_bb,
new KeyValue()); }
+  public static KeyValue getRootAsKeyValue(ByteBuffer _bb, KeyValue obj) { _bb.order(ByteOrder.LITTLE_ENDIAN);
return (obj.__assign(_bb.getInt(_bb.position()) + _bb.position(), _bb)); }
+  public void __init(int _i, ByteBuffer _bb) { bb_pos = _i; bb = _bb; }
+  public KeyValue __assign(int _i, ByteBuffer _bb) { __init(_i, _bb); return this; }
+
+  public String key() { int o = __offset(4); return o != 0 ? __string(o + bb_pos) : null;
}
+  public ByteBuffer keyAsByteBuffer() { return __vector_as_bytebuffer(4, 1); }
+  public ByteBuffer keyInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb,
4, 1); }
+  public String value() { int o = __offset(6); return o != 0 ? __string(o + bb_pos) : null;
}
+  public ByteBuffer valueAsByteBuffer() { return __vector_as_bytebuffer(6, 1); }
+  public ByteBuffer valueInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb,
6, 1); }
+
+  public static int createKeyValue(FlatBufferBuilder builder,
+      int keyOffset,
+      int valueOffset) {
+    builder.startObject(2);
+    KeyValue.addValue(builder, valueOffset);
+    KeyValue.addKey(builder, keyOffset);
+    return KeyValue.endKeyValue(builder);
+  }
+
+  public static void startKeyValue(FlatBufferBuilder builder) { builder.startObject(2); }
+  public static void addKey(FlatBufferBuilder builder, int keyOffset) { builder.addOffset(0,
keyOffset, 0); }
+  public static void addValue(FlatBufferBuilder builder, int valueOffset) { builder.addOffset(1,
valueOffset, 0); }
+  public static int endKeyValue(FlatBufferBuilder builder) {
+    int o = builder.endObject();
+    return o;
+  }
+}
+
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/Message.java
b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/Message.java
new file mode 100644
index 0000000..d29171c
--- /dev/null
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/Message.java
@@ -0,0 +1,53 @@
+// automatically generated by the FlatBuffers compiler, do not modify
+
+package org.apache.pulsar.io.kinesis.fbs;
+
+import java.nio.*;
+import java.lang.*;
+import java.util.*;
+import com.google.flatbuffers.*;
+
+@SuppressWarnings("unused")
+public final class Message extends Table {
+  public static Message getRootAsMessage(ByteBuffer _bb) { return getRootAsMessage(_bb, new
Message()); }
+  public static Message getRootAsMessage(ByteBuffer _bb, Message obj) { _bb.order(ByteOrder.LITTLE_ENDIAN);
return (obj.__assign(_bb.getInt(_bb.position()) + _bb.position(), _bb)); }
+  public void __init(int _i, ByteBuffer _bb) { bb_pos = _i; bb = _bb; }
+  public Message __assign(int _i, ByteBuffer _bb) { __init(_i, _bb); return this; }
+
+  public EncryptionCtx encryptionCtx() { return encryptionCtx(new EncryptionCtx()); }
+  public EncryptionCtx encryptionCtx(EncryptionCtx obj) { int o = __offset(4); return o !=
0 ? obj.__assign(__indirect(o + bb_pos), bb) : null; }
+  public KeyValue properties(int j) { return properties(new KeyValue(), j); }
+  public KeyValue properties(KeyValue obj, int j) { int o = __offset(6); return o != 0 ?
obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
+  public int propertiesLength() { int o = __offset(6); return o != 0 ? __vector_len(o) :
0; }
+  public byte payload(int j) { int o = __offset(8); return o != 0 ? bb.get(__vector(o) +
j * 1) : 0; }
+  public int payloadLength() { int o = __offset(8); return o != 0 ? __vector_len(o) : 0;
}
+  public ByteBuffer payloadAsByteBuffer() { return __vector_as_bytebuffer(8, 1); }
+  public ByteBuffer payloadInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb,
8, 1); }
+
+  public static int createMessage(FlatBufferBuilder builder,
+      int encryptionCtxOffset,
+      int propertiesOffset,
+      int payloadOffset) {
+    builder.startObject(3);
+    Message.addPayload(builder, payloadOffset);
+    Message.addProperties(builder, propertiesOffset);
+    Message.addEncryptionCtx(builder, encryptionCtxOffset);
+    return Message.endMessage(builder);
+  }
+
+  public static void startMessage(FlatBufferBuilder builder) { builder.startObject(3); }
+  public static void addEncryptionCtx(FlatBufferBuilder builder, int encryptionCtxOffset)
{ builder.addOffset(0, encryptionCtxOffset, 0); }
+  public static void addProperties(FlatBufferBuilder builder, int propertiesOffset) { builder.addOffset(1,
propertiesOffset, 0); }
+  public static int createPropertiesVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4,
data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]);
return builder.endVector(); }
+  public static void startPropertiesVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4,
numElems, 4); }
+  public static void addPayload(FlatBufferBuilder builder, int payloadOffset) { builder.addOffset(2,
payloadOffset, 0); }
+  public static int createPayloadVector(FlatBufferBuilder builder, byte[] data) { builder.startVector(1,
data.length, 1); for (int i = data.length - 1; i >= 0; i--) builder.addByte(data[i]); return
builder.endVector(); }
+  public static void startPayloadVector(FlatBufferBuilder builder, int numElems) { builder.startVector(1,
numElems, 1); }
+  public static int endMessage(FlatBufferBuilder builder) {
+    int o = builder.endObject();
+    return o;
+  }
+  public static void finishMessageBuffer(FlatBufferBuilder builder, int offset) { builder.finish(offset);
}
+  public static void finishSizePrefixedMessageBuffer(FlatBufferBuilder builder, int offset)
{ builder.finishSizePrefixed(offset); }
+}
+
diff --git a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/UtilsTest.java b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/UtilsTest.java
index e3f1160..b9f5c8a 100644
--- a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/UtilsTest.java
+++ b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/UtilsTest.java
@@ -20,15 +20,20 @@ package org.apache.pulsar.io.kinesis;
 
 import static java.util.Base64.getDecoder;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.Optional;
 
+import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.common.api.EncryptionContext;
 import org.apache.pulsar.common.api.EncryptionContext.EncryptionKey;
 import org.apache.pulsar.common.api.proto.PulsarApi.CompressionType;
 import org.apache.pulsar.io.core.RecordContext;
+import org.apache.pulsar.io.kinesis.fbs.KeyValue;
+import org.apache.pulsar.io.kinesis.fbs.Message;
 import org.testng.Assert;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 import org.testng.collections.Maps;
 
@@ -43,71 +48,166 @@ import lombok.ToString;
  */
 public class UtilsTest {
 
+    @DataProvider(name = "encryption")
+    public Object[][] encryptionProvider() {
+        return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+    }
+    
     @Test
     public void testJsonSerialization() throws Exception {
 
-        final String key1 = "key1";
-        final String key2 = "key2";
+        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;
 
-        // prepare encryption-ctx
-        EncryptionContext ctx = new EncryptionContext();
-        ctx.setAlgorithm(algo);
-        ctx.setBatchSize(Optional.of(10));
-        ctx.setCompressionType(CompressionType.LZ4);
-        ctx.setUncompressedMessageSize(10);
-        Map<String, EncryptionKey> keys = Maps.newHashMap();
-        EncryptionKey encKeyVal = new EncryptionKey();
-        encKeyVal.setKeyValue(key1Value.getBytes());
+        // 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");
-        encKeyVal.setMetadata(metadata1);
-        EncryptionKey encKeyVal2 = new EncryptionKey();
-        encKeyVal2.setKeyValue(key2Value.getBytes());
         Map<String, String> metadata2 = Maps.newHashMap();
         metadata2.put("version", "v2");
         metadata2.put("ckms", "cmks-2");
-        encKeyVal2.setMetadata(metadata2);
-        keys.put(key1, encKeyVal);
-        keys.put(key2, encKeyVal2);
-        ctx.setKeys(keys);
-        ctx.setMetadata(metadata1);
-        ctx.setParam(param.getBytes());
-
-        // serialize to json
-        byte[] data = "payload".getBytes();
-        Map<String, String> properties = Maps.newHashMap();
-        properties.put("prop1", "value");
-        RecordContext recordCtx = new RecordContextImpl(properties, ctx);
+        RecordContext recordCtx = createRecordContext(algo, keyNames, keyValues, param.getBytes(),
metadata1, metadata2,
+                batchSize, compressionMsgSize, properties, true);
         String json = Utils.serializeRecordToJson(recordCtx, data);
-        System.out.println(json);
 
         // deserialize from json and assert
         KinesisMessageResponse kinesisJsonResponse = deSerializeRecordFromJson(json);
         Assert.assertEquals(data, getDecoder().decode(kinesisJsonResponse.getPayloadBase64()));
         EncryptionCtx encryptionCtxDeser = kinesisJsonResponse.getEncryptionCtx();
-        Assert.assertEquals(key1Value.getBytes(), getDecoder().decode(encryptionCtxDeser.getKeysMapBase64().get(key1)));
-        Assert.assertEquals(key2Value.getBytes(), getDecoder().decode(encryptionCtxDeser.getKeysMapBase64().get(key2)));
+        Assert.assertEquals(key1Value.getBytes(),
+                getDecoder().decode(encryptionCtxDeser.getKeysMapBase64().get(keyNames[0])));
+        Assert.assertEquals(key2Value.getBytes(),
+                getDecoder().decode(encryptionCtxDeser.getKeysMapBase64().get(keyNames[1])));
         Assert.assertEquals(param.getBytes(), getDecoder().decode(encryptionCtxDeser.getEncParamBase64()));
         Assert.assertEquals(algo, encryptionCtxDeser.getAlgorithm());
-        Assert.assertEquals(metadata1, encryptionCtxDeser.getKeysMetadataMap().get(key1));
-        Assert.assertEquals(metadata2, encryptionCtxDeser.getKeysMetadataMap().get(key2));
-        Assert.assertEquals(metadata1, encryptionCtxDeser.getMetadata());
+        Assert.assertEquals(metadata1, encryptionCtxDeser.getKeysMetadataMap().get(keyNames[0]));
+        Assert.assertEquals(metadata2, encryptionCtxDeser.getKeysMetadataMap().get(keyNames[1]));
         Assert.assertEquals(properties, kinesisJsonResponse.getProperties());
 
     }
 
+    @Test(dataProvider="encryption")
+    public void testFbSerialization(boolean isEncryption) throws Exception {
+
+        final String[] keyNames = { "key1", "key2" };
+        final String param = "param";
+        final String algo = "algo";
+        int batchSize = 10;
+        int compressionMsgSize = 10;
+
+        for (int k = 0; k < 5; k++) {
+            String payloadString = RandomStringUtils.random(142342 * k, String.valueOf(System.currentTimeMillis()));
+            final String key1Value = payloadString + "test1";
+            final String key2Value = payloadString + "test2";
+            final byte[][] keyValues = { key1Value.getBytes(), key2Value.getBytes() };
+            byte[] data = payloadString.getBytes();
+            Map<String, String> properties = Maps.newHashMap();
+            properties.put("prop1", payloadString);
+            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");
+            RecordContext recordCtx = createRecordContext(algo, keyNames, keyValues, param.getBytes(),
metadata1,
+                    metadata2, batchSize, compressionMsgSize, properties, isEncryption);
+            ByteBuffer flatBuffer = Utils.serializeRecordToFlatBuffer(recordCtx, data);
+
+            Message kinesisJsonResponse = Message.getRootAsMessage(flatBuffer);
+            byte[] fbPayloadBytes = new byte[kinesisJsonResponse.payloadLength()];
+            kinesisJsonResponse.payloadAsByteBuffer().get(fbPayloadBytes);
+            Assert.assertEquals(data, fbPayloadBytes);
+
+            if(isEncryption) {
+                org.apache.pulsar.io.kinesis.fbs.EncryptionCtx encryptionCtxDeser = kinesisJsonResponse.encryptionCtx();
+                byte compressionType = encryptionCtxDeser.compressionType();
+                int fbBatchSize = encryptionCtxDeser.batchSize();
+                boolean isBathcMessage = encryptionCtxDeser.isBatchMessage();
+                int fbCompressionMsgSize = encryptionCtxDeser.uncompressedMessageSize();
+                int totalKeys = encryptionCtxDeser.keysLength();
+                Map<String, Map<String, String>> fbKeyMetadataResult = Maps.newHashMap();
+                Map<String, byte[]> fbKeyValueResult = Maps.newHashMap();
+                for (int i = 0; i < encryptionCtxDeser.keysLength(); i++) {
+                    org.apache.pulsar.io.kinesis.fbs.EncryptionKey encryptionKey = encryptionCtxDeser.keys(i);
+                    String keyName = encryptionKey.key();
+                    byte[] keyValueBytes = new byte[encryptionKey.valueLength()];
+                    encryptionKey.valueAsByteBuffer().get(keyValueBytes);
+                    fbKeyValueResult.put(keyName, keyValueBytes);
+                    Map<String, String> fbMetadata = Maps.newHashMap();
+                    for (int j = 0; j < encryptionKey.metadataLength(); j++) {
+                        KeyValue encMtdata = encryptionKey.metadata(j);
+                        fbMetadata.put(encMtdata.key(), encMtdata.value());
+                    }
+                    fbKeyMetadataResult.put(keyName, fbMetadata);
+                }
+                byte[] paramBytes = new byte[encryptionCtxDeser.paramLength()];
+                encryptionCtxDeser.paramAsByteBuffer().get(paramBytes);
+
+                Assert.assertEquals(totalKeys, 2);
+                Assert.assertEquals(batchSize, fbBatchSize);
+                Assert.assertEquals(isBathcMessage, true);
+                Assert.assertEquals(compressionMsgSize, fbCompressionMsgSize);
+                Assert.assertEquals(keyValues[0], fbKeyValueResult.get(keyNames[0]));
+                Assert.assertEquals(keyValues[1], fbKeyValueResult.get(keyNames[1]));
+                Assert.assertEquals(metadata1, fbKeyMetadataResult.get(keyNames[0]));
+                Assert.assertEquals(metadata2, fbKeyMetadataResult.get(keyNames[1]));
+                Assert.assertEquals(compressionType, org.apache.pulsar.io.kinesis.fbs.CompressionType.LZ4);
+                Assert.assertEquals(param.getBytes(), paramBytes);
+                Assert.assertEquals(algo, encryptionCtxDeser.algo());
+            }
+            
+            Map<String, String> fbproperties = Maps.newHashMap();
+            for (int i = 0; i < kinesisJsonResponse.propertiesLength(); i++) {
+                KeyValue property = kinesisJsonResponse.properties(i);
+                fbproperties.put(property.key(), property.value());
+            }
+            Assert.assertEquals(properties, fbproperties);
+
+        }
+    }
+
+    private RecordContext createRecordContext(String algo, String[] keyNames, byte[][] keyValues,
byte[] param,
+            Map<String, String> metadata1, Map<String, String> metadata2, int
batchSize, int compressionMsgSize,
+            Map<String, String> properties, boolean isEncryption) {
+        EncryptionContext ctx = null;
+        if(isEncryption) {
+            ctx = new EncryptionContext();
+            ctx.setAlgorithm(algo);
+            ctx.setBatchSize(Optional.of(batchSize));
+            ctx.setCompressionType(CompressionType.LZ4);
+            ctx.setUncompressedMessageSize(compressionMsgSize);
+            Map<String, EncryptionKey> keys = Maps.newHashMap();
+            EncryptionKey encKeyVal = new EncryptionKey();
+            encKeyVal.setKeyValue(keyValues[0]);
+
+            encKeyVal.setMetadata(metadata1);
+            EncryptionKey encKeyVal2 = new EncryptionKey();
+            encKeyVal2.setKeyValue(keyValues[1]);
+            encKeyVal2.setMetadata(metadata2);
+            keys.put(keyNames[0], encKeyVal);
+            keys.put(keyNames[1], encKeyVal2);
+            ctx.setKeys(keys);
+            ctx.setParam(param);
+        }
+        return new RecordContextImpl(properties, Optional.ofNullable(ctx)); 
+    }
+
     class RecordContextImpl implements RecordContext {
         Map<String, String> properties;
         Optional<EncryptionContext> ectx;
 
-        public RecordContextImpl(Map<String, String> properties, EncryptionContext
ectx) {
+        public RecordContextImpl(Map<String, String> properties, Optional<EncryptionContext>
ectx) {
             this.properties = properties;
-            this.ectx = Optional.of(ectx);
+            this.ectx = ectx;
         }
 
         public Map<String, String> getProperties() {
@@ -146,8 +246,6 @@ public class UtilsTest {
         private Map<String, String> keysMapBase64;
         // map of encryption-key metadata
         private Map<String, Map<String, String>> keysMetadataMap;
-        // encryption-ctx metadata
-        private Map<String, String> metadata;
         // encryption param which is base64 encoded
         private String encParamBase64;
         // encryption algorithm


Mime
View raw message