diff --git a/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializer.java b/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializer.java
index 7710b78c8a..d9cce50677 100644
--- a/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializer.java
+++ b/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializer.java
@@ -28,6 +28,9 @@ class EncryptionSerializer<T> implements Serializer<T> {
 
     @Override
     public byte[] serialize(String topic, T data) {
+        if(data == null)
+            return null;
+
         byte[] unencryptedBytes = innerSerializer.serialize(topic, data);
         try {
             return cipher.doFinal(unencryptedBytes);
diff --git a/data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializerTest.java b/data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializerTest.java
index 941b27e1ea..342db03d9f 100644
--- a/data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializerTest.java
+++ b/data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/common/serialization/EncryptionSerializerTest.java
@@ -16,7 +16,9 @@
 import java.util.UUID;
 
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
@@ -54,4 +56,12 @@ void serialize_performs_cipher_encryption_on_serialized_data() throws IllegalBlo
         assertThat(createObjectUnderTest().serialize(topicName, input),
                 equalTo(encryptedData));
     }
+
+    @Test
+    void serialize_returns_null_and_does_not_call_cipher_if_input_is_null() throws NoSuchPaddingException, NoSuchAlgorithmException, InvalidKeyException {
+        assertThat(createObjectUnderTest().serialize(topicName, null),
+                nullValue());
+
+        verifyNoInteractions(cipher);
+    }
 }
\ No newline at end of file