fix spark kafka source offset field is null

harbby · Sep 15, 2020 · 550a42b · 550a42b
1 parent 8a27b0c
commit 550a42b
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 0 deletions.
diff --git a/sylph-connectors/spark-kafka/src/main/java/ideal/sylph/plugins/kafka/spark/KafkaSource.java b/sylph-connectors/spark-kafka/src/main/java/ideal/sylph/plugins/kafka/spark/KafkaSource.java
@@ -122,10 +122,13 @@ public void commitOffsets(RDD<?> kafkaRdd)
                                     continue;
                                 case "_offset":
                                     values[i] = record.offset();
+                                    continue;
                                 case "_timestamp":
                                     values[i] = record.timestamp();
+                                    continue;
                                 case "_timestampType":
                                     values[i] = record.timestampType().id;
+                                    continue;
                                 default:
                                     values[i] = null;
                             }

diff --git a/...tors/spark-kafka/src/main/java/ideal/sylph/plugins/kafka/spark/StructuredKafkaSource.java b/...tors/spark-kafka/src/main/java/ideal/sylph/plugins/kafka/spark/StructuredKafkaSource.java
@@ -108,6 +108,7 @@ private static Dataset<Row> createSource(SparkSession spark, KafkaSourceConfig c
                                     continue;
                                 case "_offset":
                                     values[i] = record.<Long>getAs("offset");
+                                    continue;
                                 default:
                                     values[i] = null;
                             }