fix: correct lag offset

Author: Jamie Chapman-Brown

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskIOConfig.java

@@ -45,8 +45,6 @@ public RabbitStreamIndexTaskIOConfig(
                                                                   // backward
                                                                   // compabitility
       @JsonProperty("baseSequenceName") String baseSequenceName,
-      // startSequenceNumbers and endSequenceNumbers must be set for new
-      // versions
       @JsonProperty("startSequenceNumbers") SeekableStreamStartSequenceNumbers<String, Long> startSequenceNumbers,
       @JsonProperty("endSequenceNumbers") SeekableStreamEndSequenceNumbers<String, Long> endSequenceNumbers,
       @JsonProperty("consumerProperties") Map<String, Object> consumerProperties,

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java

@@ -279,7 +279,7 @@ private Map<String, Long> getRecordLagPerPartitionInLatestSequences(Map<String,
             Collectors.toMap(
                 Entry::getKey,
                 e -> e.getValue() != null
-                    ? e.getValue() - Optional.ofNullable(currentOffsets.get(e.getKey())).orElse(0L)
+                    ? e.getValue() + 1 - Optional.ofNullable(currentOffsets.get(e.getKey())).orElse(0L)
                     : 0));
   }
 
@@ -303,7 +303,7 @@ protected Map<String, Long> getRecordLagPerPartition(Map<String, Long> currentOf
             Collectors.toMap(
                 Entry::getKey,
                 e -> e.getValue() != null
-                    ? latestSequenceFromStream.get(e.getKey()) - e.getValue()
+                    ? latestSequenceFromStream.get(e.getKey()) + 1 - e.getValue()
                     : 0));
   }