Addressed issue in Kafka-pubsub for avro null messages (#3531)

Signed-off-by: Patrick Assuied <patrick.assuied@elationhealth.com>

common/component/kafka/kafka.go

@@ -14,6 +14,7 @@ limitations under the License.
 package kafka
 
 import (
+	"bytes"
 	"context"
 	"encoding/binary"
 	"errors"
@@ -263,7 +264,9 @@ func getSchemaSubject(topic string) string {
 }
 
 func (k *Kafka) DeserializeValue(message *sarama.ConsumerMessage, config SubscriptionHandlerConfig) ([]byte, error) {
-	// Null Data is valid and a tombstone record. It shouldn't be serialized
+	// Null Data is valid and a tombstone record.
+	// It shouldn't be going through schema validation and decoding
+	// Instead directly convert to JSON `null`
 	if message.Value == nil {
 		return []byte("null"), nil
 	}
@@ -354,8 +357,9 @@ func (k *Kafka) getSchemaRegistyClient() (srclient.ISchemaRegistryClient, error)
 }
 
 func (k *Kafka) SerializeValue(topic string, data []byte, metadata map[string]string) ([]byte, error) {
-	// Null Data is valid and a tombstone record. It shouldn't be serialized
-	if data == nil {
+	// Null Data is valid and a tombstone record.
+	// It should be converted to NULL and not go through schema validation & encoding
+	if bytes.Equal(data, []byte("null")) || data == nil {
 		return nil, nil
 	}
 

common/component/kafka/kafka_test.go

@@ -219,7 +219,14 @@ func TestSerializeValueCachingDisabled(t *testing.T) {
 		require.NoError(t, err)
 	})
 
-	t.Run("value published null, no error", func(t *testing.T) {
+	t.Run("value published 'null', no error", func(t *testing.T) {
+		act, err := k.SerializeValue("my-topic", []byte("null"), map[string]string{"valueSchemaType": "Avro"})
+
+		require.Nil(t, act)
+		require.NoError(t, err)
+	})
+
+	t.Run("value published nil, no error", func(t *testing.T) {
 		act, err := k.SerializeValue("my-topic", nil, map[string]string{"valueSchemaType": "Avro"})
 
 		require.Nil(t, act)