Fix kafka offset from option

Author: poorna2152

component/src/main/java/io/siddhi/extension/io/kafka/source/KafkaConsumerThread.java

@@ -290,7 +290,7 @@ public void run() {
                                     }
                                 }
                                 if (!isReplayThread) {
-                                    kafkaSourceState.getTopicOffsetMap().get(record.topic()).put(record.partition(),
+                                    kafkaSourceState.getTopicOffsetMap().get(record.topic()).putIfAbsent(record.partition(),
                                             record.offset());
                                 }
                                 if (endReplay(record)) {
@@ -300,7 +300,7 @@ public void run() {
                             }
                     } else {
                         if (!isReplayThread) {
-                            kafkaSourceState.getTopicOffsetMap().get(record.topic()).put(record.partition(),
+                            kafkaSourceState.getTopicOffsetMap().get(record.topic()).putIfAbsent(record.partition(),
                                   record.offset());
                         }
                         if (metrics != null) {
@@ -351,7 +351,13 @@ public void run() {
     void seekToRequiredOffset() {}
 
     boolean isRecordAfterStartOffset(ConsumerRecord record) {
-        return true;
+        Map<Integer, Long> partitionMap = kafkaSourceState.getTopicOffsetMap().get(record.topic());
+        if (partitionMap == null) {
+            return true;
+        }
+
+        Long offsetThreshold = partitionMap.get(record.partition());
+        return offsetThreshold == null || record.offset() > offsetThreshold;
     }
 
     boolean endReplay(ConsumerRecord record) {

component/src/main/java/io/siddhi/extension/io/kafka/source/KafkaSource.java

@@ -314,7 +314,7 @@ public StateFactory<KafkaSourceState> init(SourceEventListener sourceEventListen
                     "true"));
             topicOffsetMapConfig = optionHolder.validateAndGetStaticValue(TOPIC_OFFSET_MAP, null);
         }
-        partitions = (partitionList != null) ? partitionList.split(KafkaIOUtils.HEADER_SEPARATOR) : null;
+        partitions = (partitionList != null) ? partitionList.split(KafkaIOUtils.HEADER_SEPARATOR) : new String[]{"0"};
         topics = topicList.split(KafkaIOUtils.HEADER_SEPARATOR);
         seqEnabled = optionHolder.validateAndGetStaticValue(SEQ_ENABLED, "false").equalsIgnoreCase("true");
         optionalConfigs = optionHolder.validateAndGetStaticValue(ADAPTOR_OPTIONAL_CONFIGURATION_PROPERTIES, null);