Merge branch 'master' into FSTORE-1023

java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/hudi/DeltaStreamerKafkaSource.java

@@ -87,15 +87,15 @@ protected InputBatch<JavaRDD<GenericRecord>> fetchNewData(Option<String> lastChe
     if (totalNewMsgs <= 0L) {
       return new InputBatch(Option.empty(), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
     } else {
-      JavaRDD<GenericRecord> newDataRdd = this.toRdd(offsetRanges, props.getString(HudiEngine.SCHEMA_ID));
+      JavaRDD<GenericRecord> newDataRdd = this.toRdd(offsetRanges);
       return new InputBatch(Option.of(newDataRdd), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
     }
   }
 
-  private JavaRDD<GenericRecord> toRdd(OffsetRange[] offsetRanges, String schemaId) {
+  private JavaRDD<GenericRecord> toRdd(OffsetRange[] offsetRanges) {
     return KafkaUtils.createRDD(this.sparkContext, this.offsetGen.getKafkaParams(), offsetRanges,
         LocationStrategies.PreferConsistent())
-            .filter(obj -> schemaId.equals(getHeader(obj.headers(), "schemaId")))
+            .filter(obj -> obj.value() != null)
             .map(obj -> (GenericRecord) obj.value());
   }