[Bug]: KafkaIO missing initial messages when using automatic topic creation

### 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 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.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 partitions = new ArrayList<>();
for (int i = 0; i < 20; i++) {
partitions.add(new TopicPartition(TOPIC_NAME, i));
}

PCollection>> ticks = p.apply(
KafkaIO.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 findMissingNumbersInFiles(File folder) {
File[] files = folder.listFiles();

TreeSet numbers = new TreeSet<>();
for (File file : files) {
numbers.addAll(readNumbersFromFile(file));
}

List 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 readNumbersFromFile(File file) {
List 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
- [x] 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