[HUDI-7450] Fix invalid kafka offset range bug in KafkaOffsetGen.computeOffsetRanges

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 0.15.0, 1.0.0
• Component/s: deltastreamer
• Labels: pull-request-available
• Story Points: 4
• Assignee: Vinish Reddy
• Reporter: Vinish Reddy
• Created: 27/Feb/24 12:58
• Updated: 09/Jun/24 22:49
• Resolved: 28/Feb/24 12:42

Description

The offset computation fails with invalid offset range for the following edge case. 

 _{@Test
  public void testBug() {
    int[] partitions = new int[] {0, 1, 2, 3, 4};
    long[] committedOffsets =
        new long[] {76888767, 76725043, 76899767, 76833267, 76952055};
    long[] latestOffsets =
        new long[] {77005407, 76768151, 76985456, 76917973, 77080447};
    OffsetRange[] ranges =
        KafkaOffsetGen.CheckpointUtils.computeOffsetRanges(
            makeOffsetMap(partitions, committedOffsets),
            makeOffsetMap(partitions, latestOffsets),
            400000,
            20);}

    _{long totalNewMsgs = KafkaOffsetGen.CheckpointUtils.totalNewMessages(ranges);
    assertEquals(400000, totalNewMsgs);
    for (OffsetRange range : ranges) {
      if (range.fromOffset() > range.untilOffset()) {
        throw new IllegalArgumentException("Invalid offset range " + range);
      }
    }
  }}

Issue Links

• links to: GitHub Pull Request #10768
• links to: GitHub Pull Request #10771