Check if put.getSchemaId is positive, if not default to 1. Current re…

…cord is a chunk. We only write to the storage engine for fully assembled records.

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -3083,7 +3083,8 @@ private int processKafkaDataMessage(
         // update checksum for this PUT message if needed.
         partitionConsumptionState.maybeUpdateExpectedChecksum(keyBytes, put);
 
-        int putSchemaId = put.getSchemaId();
+        // Check if put.getSchemaId is positive, if not default to 1
+        int putSchemaId = put.getSchemaId() > 0 ? put.getSchemaId() : 1;
 
         // Do transformation recompute key, value and partition
         if (recordTransformer != null) {
@@ -3101,6 +3102,11 @@ private int processKafkaDataMessage(
               putSchemaId,
               compressorFactory.getCompressor(compressionStrategy, kafkaVersionTopic));
 
+          // Current record is a chunk. We only write to the storage engine for fully assembled records
+          if (assembledObject == null) {
+            break;
+          }
+
           SchemaEntry keySchema = schemaRepository.getKeySchema(storeName);
           Lazy<Object> lazyKey = Lazy.of(() -> deserializeAvroObjectAndReturn(ByteBuffer.wrap(keyBytes), keySchema));
           Lazy<Object> lazyValue = Lazy.of(() -> assembledObject);