You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/05/14 21:39:14 UTC
(hudi) 07/13: [HUDI-7506] Compute offsetRanges based on eventsPerPartition allocated in each range (#10869)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit ab9513b61991c860799e6ad877f6bea25c5bc52f
Author: Vinish Reddy <vi...@gmail.com>
AuthorDate: Sat Mar 16 00:50:53 2024 +0530
[HUDI-7506] Compute offsetRanges based on eventsPerPartition allocated in each range (#10869)
---
.../utilities/sources/helpers/KafkaOffsetGen.java | 88 ++++++-----
.../sources/helpers/TestCheckpointUtils.java | 167 ++++++++++++++++-----
.../sources/helpers/TestKafkaOffsetGen.java | 10 +-
3 files changed, 179 insertions(+), 86 deletions(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
index 9b1f8674ca8..442046cd948 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
@@ -41,10 +41,10 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -114,6 +114,7 @@ public class KafkaOffsetGen {
* @param fromOffsetMap offsets where we left off last time
* @param toOffsetMap offsets of where each partitions is currently at
* @param numEvents maximum number of events to read.
+ * @param minPartitions minimum partitions used for
*/
public static OffsetRange[] computeOffsetRanges(Map<TopicPartition, Long> fromOffsetMap,
Map<TopicPartition, Long> toOffsetMap,
@@ -129,63 +130,58 @@ public class KafkaOffsetGen {
.toArray(new OffsetRange[toOffsetMap.size()]);
LOG.debug("numEvents {}, minPartitions {}, ranges {}", numEvents, minPartitions, ranges);
- boolean needSplitToMinPartitions = minPartitions > toOffsetMap.size();
- long totalEvents = totalNewMessages(ranges);
- long allocatedEvents = 0;
- Set<Integer> exhaustedPartitions = new HashSet<>();
- List<OffsetRange> finalRanges = new ArrayList<>();
// choose the actualNumEvents with min(totalEvents, numEvents)
- long actualNumEvents = Math.min(totalEvents, numEvents);
-
- // keep going until we have events to allocate and partitions still not exhausted.
- while (allocatedEvents < numEvents && exhaustedPartitions.size() < toOffsetMap.size()) {
- // Allocate the remaining events to non-exhausted partitions, in round robin fashion
- Set<Integer> allocatedPartitionsThisLoop = new HashSet<>(exhaustedPartitions);
- for (int i = 0; i < ranges.length; i++) {
- long remainingEvents = actualNumEvents - allocatedEvents;
- long remainingPartitions = toOffsetMap.size() - allocatedPartitionsThisLoop.size();
- // if need tp split into minPartitions, recalculate the remainingPartitions
- if (needSplitToMinPartitions) {
- remainingPartitions = minPartitions - finalRanges.size();
+ long actualNumEvents = Math.min(totalNewMessages(ranges), numEvents);
+ minPartitions = Math.max(minPartitions, toOffsetMap.size());
+ // Each OffsetRange computed will have maximum of eventsPerPartition,
+ // this ensures all ranges are evenly distributed and there's no skew in one particular range.
+ long eventsPerPartition = Math.max(1L, actualNumEvents / minPartitions);
+ long allocatedEvents = 0;
+ Map<TopicPartition, List<OffsetRange>> finalRanges = new HashMap<>();
+ Map<TopicPartition, Long> partitionToAllocatedOffset = new HashMap<>();
+ // keep going until we have events to allocate.
+ while (allocatedEvents < actualNumEvents) {
+ // Allocate the remaining events in round-robin fashion.
+ for (OffsetRange range : ranges) {
+ // if we have already allocated required no of events, exit
+ if (allocatedEvents == actualNumEvents) {
+ break;
}
- long eventsPerPartition = (long) Math.ceil((1.0 * remainingEvents) / remainingPartitions);
-
- OffsetRange range = ranges[i];
- if (exhaustedPartitions.contains(range.partition())) {
- continue;
+ // Compute startOffset.
+ long startOffset = range.fromOffset();
+ if (partitionToAllocatedOffset.containsKey(range.topicPartition())) {
+ startOffset = partitionToAllocatedOffset.get(range.topicPartition());
}
-
+ // for last bucket, we may not have full eventsPerPartition msgs.
+ long eventsForThisPartition = Math.min(eventsPerPartition, (actualNumEvents - allocatedEvents));
+ // Compute toOffset.
long toOffset = -1L;
- if (range.fromOffset() + eventsPerPartition > range.fromOffset()) {
- toOffset = Math.min(range.untilOffset(), range.fromOffset() + eventsPerPartition);
+ if (startOffset + eventsForThisPartition > startOffset) {
+ toOffset = Math.min(range.untilOffset(), startOffset + eventsForThisPartition);
} else {
// handling Long overflow
toOffset = range.untilOffset();
}
- if (toOffset == range.untilOffset()) {
- exhaustedPartitions.add(range.partition());
- }
- // We need recompute toOffset if we have allocatedEvents are more than actualNumEvents.
- long totalAllocatedEvents = allocatedEvents + (toOffset - range.fromOffset());
- if (totalAllocatedEvents > actualNumEvents) {
- long offsetsToAdd = Math.min(eventsPerPartition, (actualNumEvents - allocatedEvents));
- toOffset = Math.min(range.untilOffset(), range.fromOffset() + offsetsToAdd);
+ allocatedEvents += toOffset - startOffset;
+ OffsetRange thisRange = OffsetRange.create(range.topicPartition(), startOffset, toOffset);
+ // Add the offsetRange(startOffset,toOffset) to finalRanges.
+ if (!finalRanges.containsKey(range.topicPartition())) {
+ finalRanges.put(range.topicPartition(), new ArrayList<>(Collections.singleton(thisRange)));
+ partitionToAllocatedOffset.put(range.topicPartition(), thisRange.untilOffset());
+ } else if (toOffset > startOffset) {
+ finalRanges.get(range.topicPartition()).add(thisRange);
+ partitionToAllocatedOffset.put(range.topicPartition(), thisRange.untilOffset());
}
- allocatedEvents += toOffset - range.fromOffset();
- OffsetRange thisRange = OffsetRange.create(range.topicPartition(), range.fromOffset(), toOffset);
- finalRanges.add(thisRange);
- ranges[i] = OffsetRange.create(range.topicPartition(), range.fromOffset() + thisRange.count(), range.untilOffset());
- allocatedPartitionsThisLoop.add(range.partition());
}
}
-
- if (!needSplitToMinPartitions) {
- LOG.debug("final ranges merged by topic partition {}", Arrays.toString(mergeRangesByTopicPartition(finalRanges.toArray(new OffsetRange[0]))));
- return mergeRangesByTopicPartition(finalRanges.toArray(new OffsetRange[0]));
+ OffsetRange[] sortedRangeArray = finalRanges.values().stream().flatMap(Collection::stream)
+ .sorted(SORT_BY_PARTITION).toArray(OffsetRange[]::new);
+ if (actualNumEvents == 0) {
+ // We return the same ranges back in case of 0 events for checkpoint computation.
+ sortedRangeArray = ranges;
}
- finalRanges.sort(SORT_BY_PARTITION);
- LOG.debug("final ranges {}", Arrays.toString(finalRanges.toArray(new OffsetRange[0])));
- return finalRanges.toArray(new OffsetRange[0]);
+ LOG.info("final ranges {}", Arrays.toString(sortedRangeArray));
+ return sortedRangeArray;
}
/**
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCheckpointUtils.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCheckpointUtils.java
index b77fb15803f..7e8b263de33 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCheckpointUtils.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCheckpointUtils.java
@@ -24,6 +24,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.spark.streaming.kafka010.OffsetRange;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -104,26 +105,35 @@ public class TestCheckpointUtils {
ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0, 1}, new long[] {200000, 250000}),
makeOffsetMap(new int[] {0, 1, 2}, new long[] {200010, 350000, 10000}), 100000, 0);
assertEquals(100000, CheckpointUtils.totalNewMessages(ranges));
+ assertEquals(5, ranges.length);
+ assertEquals(0, ranges[0].partition());
assertEquals(10, ranges[0].count());
- assertEquals(89990, ranges[1].count());
- assertEquals(10000, ranges[2].count());
+ assertEquals(1, ranges[1].partition());
+ assertEquals(33333, ranges[1].count());
+ assertEquals(33333, ranges[2].count());
+ assertEquals(23324, ranges[3].count());
+ assertEquals(2, ranges[4].partition());
+ assertEquals(10000, ranges[4].count());
ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0, 1}, new long[] {200000, 250000}),
makeOffsetMap(new int[] {0, 1, 2}, new long[] {200010, 350000, 10000}), 1000000, 0);
assertEquals(110010, CheckpointUtils.totalNewMessages(ranges));
assertEquals(10, ranges[0].count());
- assertEquals(100000, ranges[1].count());
- assertEquals(10000, ranges[2].count());
+ assertEquals(36670, ranges[1].count());
+ assertEquals(36670, ranges[2].count());
+ assertEquals(26660, ranges[3].count());
+ assertEquals(10000, ranges[4].count());
// not all partitions consume same entries.
ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0, 1, 2, 3, 4}, new long[] {0, 0, 0, 0, 0}),
makeOffsetMap(new int[] {0, 1, 2, 3, 4}, new long[] {100, 1000, 1000, 1000, 1000}), 1001, 0);
assertEquals(1001, CheckpointUtils.totalNewMessages(ranges));
assertEquals(100, ranges[0].count());
- assertEquals(226, ranges[1].count());
- assertEquals(225, ranges[2].count());
- assertEquals(225, ranges[3].count());
- assertEquals(225, ranges[4].count());
+ assertEquals(200, ranges[1].count());
+ assertEquals(101, ranges[2].count());
+ assertEquals(200, ranges[3].count());
+ assertEquals(200, ranges[4].count());
+ assertEquals(200, ranges[5].count());
}
@Test
@@ -167,38 +177,44 @@ public class TestCheckpointUtils {
// N skewed TopicPartitions to M offset ranges
ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0, 1}, new long[] {0, 0}),
makeOffsetMap(new int[] {0, 1}, new long[] {100, 500}), 600, 3);
- assertEquals(3, ranges.length);
+ assertEquals(4, ranges.length);
assertEquals(0, ranges[0].fromOffset());
assertEquals(100, ranges[0].untilOffset());
assertEquals(0, ranges[1].fromOffset());
- assertEquals(250, ranges[1].untilOffset());
- assertEquals(250, ranges[2].fromOffset());
- assertEquals(500, ranges[2].untilOffset());
+ assertEquals(200, ranges[1].untilOffset());
+ assertEquals(200, ranges[2].fromOffset());
+ assertEquals(400, ranges[2].untilOffset());
+ assertEquals(400, ranges[3].fromOffset());
+ assertEquals(500, ranges[3].untilOffset());
// range inexact multiple of minPartitions
ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0}, new long[] {0}),
makeOffsetMap(new int[] {0}, new long[] {100}), 600, 3);
- assertEquals(3, ranges.length);
+ assertEquals(4, ranges.length);
assertEquals(0, ranges[0].fromOffset());
- assertEquals(34, ranges[0].untilOffset());
- assertEquals(34, ranges[1].fromOffset());
- assertEquals(67, ranges[1].untilOffset());
- assertEquals(67, ranges[2].fromOffset());
- assertEquals(100, ranges[2].untilOffset());
+ assertEquals(33, ranges[0].untilOffset());
+ assertEquals(33, ranges[1].fromOffset());
+ assertEquals(66, ranges[1].untilOffset());
+ assertEquals(66, ranges[2].fromOffset());
+ assertEquals(99, ranges[2].untilOffset());
+ assertEquals(99, ranges[3].fromOffset());
+ assertEquals(100, ranges[3].untilOffset());
// do not ignore empty ranges
ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0, 1}, new long[] {100, 0}),
makeOffsetMap(new int[] {0, 1}, new long[] {100, 600}), 600, 3);
- assertEquals(3, ranges.length);
+ assertEquals(4, ranges.length);
assertEquals(0, ranges[0].partition());
assertEquals(100, ranges[0].fromOffset());
assertEquals(100, ranges[0].untilOffset());
assertEquals(1, ranges[1].partition());
assertEquals(0, ranges[1].fromOffset());
- assertEquals(300, ranges[1].untilOffset());
+ assertEquals(200, ranges[1].untilOffset());
assertEquals(1, ranges[2].partition());
- assertEquals(300, ranges[2].fromOffset());
- assertEquals(600, ranges[2].untilOffset());
+ assertEquals(200, ranges[2].fromOffset());
+ assertEquals(400, ranges[2].untilOffset());
+ assertEquals(400, ranges[3].fromOffset());
+ assertEquals(600, ranges[3].untilOffset());
// all empty ranges, do not ignore empty ranges
ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0, 1}, new long[] {100, 0}),
@@ -227,7 +243,7 @@ public class TestCheckpointUtils {
OffsetRange range = OffsetRange.apply(TEST_TOPIC_NAME, 0, 0, 100);
OffsetRange[] ranges = CheckpointUtils.computeOffsetRanges(makeOffsetMap(new int[] {0, 1}, new long[] {0, 0}),
makeOffsetMap(new int[] {0, 1}, new long[] {100, 500}), 600, 4);
- assertEquals(4, ranges.length);
+ assertEquals(5, ranges.length);
OffsetRange[] mergedRanges = CheckpointUtils.mergeRangesByTopicPartition(ranges);
assertEquals(2, mergedRanges.length);
assertEquals(0, mergedRanges[0].partition());
@@ -253,12 +269,14 @@ public class TestCheckpointUtils {
new long[] {76888767, 76725043, 76899767, 76833267, 76952055};
long[] latestOffsets =
new long[] {77005407, 76768151, 76985456, 76917973, 77080447};
+ long numEvents = 400000;
+ long minPartitions = 20;
OffsetRange[] ranges =
KafkaOffsetGen.CheckpointUtils.computeOffsetRanges(
makeOffsetMap(partitions, committedOffsets),
makeOffsetMap(partitions, latestOffsets),
- 400000,
- 20);
+ numEvents,
+ minPartitions);
long totalNewMsgs = KafkaOffsetGen.CheckpointUtils.totalNewMessages(ranges);
assertEquals(400000, totalNewMsgs);
@@ -267,30 +285,107 @@ public class TestCheckpointUtils {
throw new IllegalArgumentException("Invalid offset range " + range);
}
}
+ long eventPerPartition = numEvents / minPartitions;
+ long rangesWhereDiffIsLessThanEventsPerPartition = Arrays.stream(ranges).filter(offsetRange -> offsetRange.untilOffset() - offsetRange.fromOffset() <= eventPerPartition).count();
+ assertEquals(ranges.length, rangesWhereDiffIsLessThanEventsPerPartition);
OffsetRange[] expectedRanges = new OffsetRange[] {
OffsetRange.apply(TEST_TOPIC_NAME, 0, 76888767, 76908767),
OffsetRange.apply(TEST_TOPIC_NAME, 0, 76908767, 76928767),
OffsetRange.apply(TEST_TOPIC_NAME, 0, 76928767, 76948767),
- OffsetRange.apply(TEST_TOPIC_NAME, 0, 76948767, 76970879),
- OffsetRange.apply(TEST_TOPIC_NAME, 0, 76970879, 76992990),
+ OffsetRange.apply(TEST_TOPIC_NAME, 0, 76948767, 76968767),
+ OffsetRange.apply(TEST_TOPIC_NAME, 0, 76968767, 76988767),
OffsetRange.apply(TEST_TOPIC_NAME, 1, 76725043, 76745043),
OffsetRange.apply(TEST_TOPIC_NAME, 1, 76745043, 76765043),
OffsetRange.apply(TEST_TOPIC_NAME, 1, 76765043, 76768151),
OffsetRange.apply(TEST_TOPIC_NAME, 2, 76899767, 76919767),
OffsetRange.apply(TEST_TOPIC_NAME, 2, 76919767, 76939767),
- OffsetRange.apply(TEST_TOPIC_NAME, 2, 76939767, 76961879),
- OffsetRange.apply(TEST_TOPIC_NAME, 2, 76961879, 76983990),
- OffsetRange.apply(TEST_TOPIC_NAME, 2, 76983990, 76983990),
+ OffsetRange.apply(TEST_TOPIC_NAME, 2, 76939767, 76959767),
+ OffsetRange.apply(TEST_TOPIC_NAME, 2, 76959767, 76979767),
+ OffsetRange.apply(TEST_TOPIC_NAME, 2, 76979767, 76985456),
OffsetRange.apply(TEST_TOPIC_NAME, 3, 76833267, 76853267),
OffsetRange.apply(TEST_TOPIC_NAME, 3, 76853267, 76873267),
- OffsetRange.apply(TEST_TOPIC_NAME, 3, 76873267, 76895379),
- OffsetRange.apply(TEST_TOPIC_NAME, 3, 76895379, 76917490),
- OffsetRange.apply(TEST_TOPIC_NAME, 3, 76917490, 76917490),
+ OffsetRange.apply(TEST_TOPIC_NAME, 3, 76873267, 76893267),
+ OffsetRange.apply(TEST_TOPIC_NAME, 3, 76893267, 76913267),
+ OffsetRange.apply(TEST_TOPIC_NAME, 3, 76913267, 76917973),
OffsetRange.apply(TEST_TOPIC_NAME, 4, 76952055, 76972055),
OffsetRange.apply(TEST_TOPIC_NAME, 4, 76972055, 76992055),
- OffsetRange.apply(TEST_TOPIC_NAME, 4, 76992055, 77014167),
- OffsetRange.apply(TEST_TOPIC_NAME, 4, 77014167, 77036278),
- OffsetRange.apply(TEST_TOPIC_NAME, 4, 77036278, 77036278),
+ OffsetRange.apply(TEST_TOPIC_NAME, 4, 76992055, 77012055),
+ OffsetRange.apply(TEST_TOPIC_NAME, 4, 77012055, 77032055),
+ OffsetRange.apply(TEST_TOPIC_NAME, 4, 77032055, 77038552),
+ };
+ assertArrayEquals(expectedRanges, ranges);
+ }
+
+ @Test
+ public void testNumAllocatedEventsLesserThanNumActualEvents() {
+ int[] partitions = new int[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17};
+ long[] committedOffsets =
+ new long[] {788543084, 787746335, 788016034, 788171708, 788327954, 788055939, 788179691, 788004145, 788105897, 788496138, 788317057, 788325907, 788287519, 787958075, 788403560, 788118894,
+ 788383733, 787273821};
+ long[] latestOffsets =
+ new long[] {788946534, 788442557, 788712188, 788867819, 789023943, 788752030, 788875648, 788700234, 788802091, 789192155, 789013192, 789021874, 788983544, 788654092, 789099516, 788814985,
+ 789079650, 787273821};
+ long numEvents = 10000000;
+ long minPartitions = 36;
+
+ OffsetRange[] ranges =
+ KafkaOffsetGen.CheckpointUtils.computeOffsetRanges(
+ makeOffsetMap(partitions, committedOffsets),
+ makeOffsetMap(partitions, latestOffsets),
+ numEvents,
+ minPartitions);
+ for (OffsetRange range : ranges) {
+ if (range.fromOffset() > range.untilOffset()) {
+ throw new IllegalArgumentException("Invalid offset range " + range);
+ }
+ }
+ assertEquals(10000000, KafkaOffsetGen.CheckpointUtils.totalNewMessages(ranges));
+ assertEquals(41, ranges.length);
+ long eventPerPartition = numEvents / minPartitions;
+ long rangesWhereDiffIsLessThanEventsPerPartition = Arrays.stream(ranges).filter(offsetRange -> offsetRange.untilOffset() - offsetRange.fromOffset() <= eventPerPartition).count();
+ assertEquals(ranges.length, rangesWhereDiffIsLessThanEventsPerPartition);
+ OffsetRange[] expectedRanges = new OffsetRange[] {
+ OffsetRange.apply(TEST_TOPIC_NAME, 0, 788543084, 788820861),
+ OffsetRange.apply(TEST_TOPIC_NAME, 0, 788820861, 788946534),
+ OffsetRange.apply(TEST_TOPIC_NAME, 1, 787746335, 788024112),
+ OffsetRange.apply(TEST_TOPIC_NAME, 1, 788024112, 788301889),
+ OffsetRange.apply(TEST_TOPIC_NAME, 1, 788301889, 788442557),
+ OffsetRange.apply(TEST_TOPIC_NAME, 2, 788016034, 788293811),
+ OffsetRange.apply(TEST_TOPIC_NAME, 2, 788293811, 788571588),
+ OffsetRange.apply(TEST_TOPIC_NAME, 2, 788571588, 788712188),
+ OffsetRange.apply(TEST_TOPIC_NAME, 3, 788171708, 788449485),
+ OffsetRange.apply(TEST_TOPIC_NAME, 3, 788449485, 788727262),
+ OffsetRange.apply(TEST_TOPIC_NAME, 3, 788727262, 788867819),
+ OffsetRange.apply(TEST_TOPIC_NAME, 4, 788327954, 788605731),
+ OffsetRange.apply(TEST_TOPIC_NAME, 4, 788605731, 788883508),
+ OffsetRange.apply(TEST_TOPIC_NAME, 4, 788883508, 789023943),
+ OffsetRange.apply(TEST_TOPIC_NAME, 5, 788055939, 788333716),
+ OffsetRange.apply(TEST_TOPIC_NAME, 5, 788333716, 788611493),
+ OffsetRange.apply(TEST_TOPIC_NAME, 5, 788611493, 788752030),
+ OffsetRange.apply(TEST_TOPIC_NAME, 6, 788179691, 788457468),
+ OffsetRange.apply(TEST_TOPIC_NAME, 6, 788457468, 788735245),
+ OffsetRange.apply(TEST_TOPIC_NAME, 6, 788735245, 788740134),
+ OffsetRange.apply(TEST_TOPIC_NAME, 7, 788004145, 788281922),
+ OffsetRange.apply(TEST_TOPIC_NAME, 7, 788281922, 788559699),
+ OffsetRange.apply(TEST_TOPIC_NAME, 8, 788105897, 788383674),
+ OffsetRange.apply(TEST_TOPIC_NAME, 8, 788383674, 788661451),
+ OffsetRange.apply(TEST_TOPIC_NAME, 9, 788496138, 788773915),
+ OffsetRange.apply(TEST_TOPIC_NAME, 9, 788773915, 789051692),
+ OffsetRange.apply(TEST_TOPIC_NAME, 10, 788317057, 788594834),
+ OffsetRange.apply(TEST_TOPIC_NAME, 10, 788594834, 788872611),
+ OffsetRange.apply(TEST_TOPIC_NAME, 11, 788325907, 788603684),
+ OffsetRange.apply(TEST_TOPIC_NAME, 11, 788603684, 788881461),
+ OffsetRange.apply(TEST_TOPIC_NAME, 12, 788287519, 788565296),
+ OffsetRange.apply(TEST_TOPIC_NAME, 12, 788565296, 788843073),
+ OffsetRange.apply(TEST_TOPIC_NAME, 13, 787958075, 788235852),
+ OffsetRange.apply(TEST_TOPIC_NAME, 13, 788235852, 788513629),
+ OffsetRange.apply(TEST_TOPIC_NAME, 14, 788403560, 788681337),
+ OffsetRange.apply(TEST_TOPIC_NAME, 14, 788681337, 788959114),
+ OffsetRange.apply(TEST_TOPIC_NAME, 15, 788118894, 788396671),
+ OffsetRange.apply(TEST_TOPIC_NAME, 15, 788396671, 788674448),
+ OffsetRange.apply(TEST_TOPIC_NAME, 16, 788383733, 788661510),
+ OffsetRange.apply(TEST_TOPIC_NAME, 16, 788661510, 788939287),
+ OffsetRange.apply(TEST_TOPIC_NAME, 17, 787273821, 787273821),
};
assertArrayEquals(expectedRanges, ranges);
}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java
index 6ad6a4c09db..d3031729e6e 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java
@@ -140,11 +140,13 @@ public class TestKafkaOffsetGen {
testUtils.sendMessages(testTopicName, Helpers.jsonifyRecords(dataGenerator.generateInserts("000", 1000)));
KafkaOffsetGen kafkaOffsetGen = new KafkaOffsetGen(getConsumerConfigs("earliest", "string"));
OffsetRange[] nextOffsetRanges = kafkaOffsetGen.getNextOffsetRanges(Option.empty(), 499, metrics);
- assertEquals(2, nextOffsetRanges.length);
+ assertEquals(3, nextOffsetRanges.length);
assertEquals(0, nextOffsetRanges[0].fromOffset());
- assertEquals(250, nextOffsetRanges[0].untilOffset());
- assertEquals(0, nextOffsetRanges[1].fromOffset());
- assertEquals(249, nextOffsetRanges[1].untilOffset());
+ assertEquals(249, nextOffsetRanges[0].untilOffset());
+ assertEquals(249, nextOffsetRanges[1].fromOffset());
+ assertEquals(250, nextOffsetRanges[1].untilOffset());
+ assertEquals(0, nextOffsetRanges[2].fromOffset());
+ assertEquals(249, nextOffsetRanges[2].untilOffset());
}
@Test