bvolpato opened a new issue, #27246:
URL: https://github.com/apache/beam/issues/27246
### What happened?
I noticed missing sequences when using `GenerateSequence + KafkaIO` to
produce records to Kafka, when relying on the automatic topic creation feature
(auto.create.topics.enable).
What I've used to reproduce:
```
Pipeline p = Pipeline.create(pipelineOptions);
PCollection<Long> numbers =
p.apply(GenerateSequence.from(0).withRate(100,
Duration.millis(100)));
numbers
.apply(
MapElements.into(
TypeDescriptors.kvs(TypeDescriptors.longs(),
TypeDescriptors.strings()))
.via(x -> KV.of(x, String.valueOf(x))))
.apply(
KafkaIO.<Long, String>write()
.withTopic(TOPIC_NAME)
.withBootstrapServers(BOOTSTRAP_SERVERS)
.withKeySerializer(LongSerializer.class)
.withValueSerializer(StringSerializer.class));
p.run();
```
And consuming using KafkaIO.read() + writing to a file:
```
Pipeline p = Pipeline.create(pipelineOptions);
List<TopicPartition> partitions = new ArrayList<>();
for (int i = 0; i < 20; i++) {
partitions.add(new TopicPartition(TOPIC_NAME, i));
}
PCollection<KV<Long, KV<String, Tick>>> ticks = p.apply(
KafkaIO.<Long, String>read()
.withBootstrapServers(BOOTSTRAP_SERVERS)
.withKeyDeserializer(LongDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
.withTopicPartitions(partitions))
.apply(MapElements.into(TypeDescriptors.strings()).via(record ->
record.getKV().getValue()));
ticks.apply("Window",
Window.into(FixedWindows.of(Duration.standardMinutes(1))))
.apply(
TextIO.write().to("gs://test-bucket/" + TOPIC_NAME + "/keys-")
.withNumShards(0)
.withWindowedWrites());
p.run();
```
Then I copy the files from gcs (`gsutil -m cp gs://test-bucket/{TOPIC}/*
/tmp/test`) and created a simple class to check for continuity:
```
public class TestContiguous {
public static void main(String[] args) {
findMissingNumbersInFiles(new File("/tmp/example-stream/"));
}
public static List<Integer> findMissingNumbersInFiles(File folder) {
File[] files = folder.listFiles();
TreeSet<Integer> numbers = new TreeSet<>();
for (File file : files) {
numbers.addAll(readNumbersFromFile(file));
}
List<Integer> missingNumbers = new ArrayList<>();
int minNumber = numbers.first();
int maxNumber = numbers.last();
for (int i = minNumber; i <= maxNumber; i++) {
if (!numbers.contains(i)) {
missingNumbers.add(i);
}
}
System.out.println(
"Min: " + minNumber + ", Max: " + maxNumber + ". Missing: " +
missingNumbers);
return missingNumbers;
}
public static List<Integer> readNumbersFromFile(File file) {
List<Integer> numbers = new ArrayList<>();
try (Scanner scanner = new Scanner(file)) {
while (scanner.hasNextLine()) {
String line = scanner.nextLine().trim();
if (!line.isEmpty()) {
try {
int number = Integer.parseInt(line.split(",")[0]);
numbers.add(number);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
return numbers;
}
}
```
It is expected that there will be some records missing at the tail end due
to parallelism, but I've always encountered about ~1k of sequences missing at
the beginning of the pipeline:
```
Min: 1268, Max: 780327. Missing: [1269, 1270, 1271, 1272, 1273, 1276, 1277,
1280, 1282, 1283, 1284, 1293, 780319, 780322, 780323, 780324, 780326]
```
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
