Aiven-Open / Aiven-Open/cloud-storage-connectors-for-apache-kafka

S3SourceConnector loses offsets on JVM restart due to Map.of() non-deterministic iteration order

Abierto
#599 1 comentario 0 reacciones 1 asignado Reclamado por @aindriu-aiven Ver en GitHub
bug
Lenguaje dominante
Java
Estrellas
58
Forks
39
Merge medio
2 d 10 h
PR fusionados (30 d)
5

Descripción

## Bug Description

The `S3SourceConnector` restarts processing from the beginning after every Kafka Connect cluster/JVM restart, even when offsets were previously committed to the `connect-offsets` topic. The connector loses all progress and reprocesses already-consumed files.

## Root Cause

`S3OffsetManagerEntry.java` builds the partition key using `Map.of()` in two places:

```java
// asKey() static method
public static OffsetManager.OffsetManagerKey asKey(final String bucket, final String s3ObjectKey) {
return () -> Map.of(BUCKET, bucket, OBJECT_KEY, s3ObjectKey);
}

// getManagerKey() instance method
@Override
public OffsetManager.OffsetManagerKey getManagerKey() {
return () -> Map.of(BUCKET, bucket, OBJECT_KEY, objectKey);
}
```

`Map.of()` returns an `ImmutableCollections.MapN` whose **iteration order is intentionally randomized at JVM startup** via a `System.nanoTime()` seed:

```java
// OpenJDK ImmutableCollections.java
static {
long seed = System.nanoTime(); // random on each JVM start
SALT32L = (int)((color * seed) >> 16) & 0xFFFF_FFFFL;
REVERSE = (SALT32L & 1) == 0; // 50/50 flip on each restart
}
```

`REVERSE` controls the direction of iteration — for a 2-entry map like `Map.of(BUCKET, ..., OBJECT_KEY, ...)`, there is a **~50% chance on each JVM restart** that the two fields are serialized in a different order.

## Failure Chain

```
JVM restart
→ new SALT32L + REVERSE randomized by System.nanoTime()
→ Map.of("bucket", ..., "objectKey", ...) iterates fields in different order
→ JsonConverter serializes the Map to different JSON bytes
JVM run 1: {"objectKey":"PATH/HERE","bucket":"bucket-name"}
JVM run 2: {"bucket":"bucket-name","objectKey":"PATH/HERE"}
→ OffsetStorageReaderImpl does ByteBuffer byte-level lookup
→ lookup misses the previously stored offset key
→ connector treats all files as unprocessed
→ starts consuming from the very first object in the bucket
```

This was confirmed by inspecting the `connect-offsets` Kafka topic directly: two different key formats for the same logical partition were observed, one written before the restart and one after.

## Impact

- Any cluster restart (planned or unplanned) causes the connector to reprocess all objects from the beginning of the bucket prefix.
- Data duplication in the destination topic proportional to the total number of objects in the bucket.
- No data loss, but potentially severe reprocessing overhead and downstream side effects for non-idempotent consumers.

## Reproduction

1. Configure `S3SourceConnector` with a bucket containing multiple objects.
2. Let it run and commit offsets (wait for at least one `offset.flush.interval.ms` cycle).
3. Restart the Kafka Connect cluster (JVM restart).
4. Observe: the connector begins reprocessing from the first object instead of resuming.
5. Inspect `connect-offsets` topic: the key written after the restart has fields in a different JSON order than the key written before.

This is non-deterministic (~50% per restart) due to the randomized `REVERSE` flag.

## Fix

Replace `Map.of()` with `LinkedHashMap` (insertion-ordered, deterministic across JVM restarts) in both `asKey()` and `getManagerKey()`:

```java
public static OffsetManager.OffsetManagerKey asKey(final String bucket, final String s3ObjectKey) {
return () -> {
final Map m = new LinkedHashMap<>();
m.put(BUCKET, bucket);
m.put(OBJECT_KEY, s3ObjectKey);
return m;
};
}

@Override
public OffsetManager.OffsetManagerKey getManagerKey() {
return () -> {
final Map m = new LinkedHashMap<>();
m.put(BUCKET, bucket);
m.put(OBJECT_KEY, objectKey);
return m;
};
}
```

Alternatively, a `TreeMap` (sorted by key name) would also work and would additionally guarantee alphabetical ordering regardless of insertion order.

## Note on Existing Deployments

Users who have already been affected by this bug will have two different key formats coexisting in their `connect-offsets` topic. After applying the fix, the connector will consistently use one format — but existing offsets stored under the old (opposite) format will remain unreachable. Affected users will need to either:
- Manually rewrite the stale offset entries to the new key format, or
- Accept a one-time reprocessing from the beginning after upgrading.

## Environment

- Connector: `io.aiven.kafka.connect.s3.source.S3SourceConnector`
- Storage backend: GCS via S3-compatible API (`https://storage.googleapis.com`)
- Kafka Connect: Strimzi-managed cluster
- Java: OpenJDK (any modern version with `ImmutableCollections.REVERSE` randomization — Java 9+)

## References

- [`S3OffsetManagerEntry.java`](https://github.com/Aiven-Open/cloud-storage-connectors-for-apache-kafka/blob/main/s3-source-connector/src/main/java/io/aiven/kafka/connect/s3/source/utils/S3OffsetManagerEntry.java)
- [`OffsetManager.java`](https://github.com/Aiven-Open/cloud-storage-connectors-for-apache-kafka/blob/main/commons/src/main/java/io/aiven/kafka/connect/common/source/OffsetManager.java)
- [OpenJDK `ImmutableCollections.java` — `SALT32L`/`REVERSE` randomization](https://github.com/openjdk/jdk/blob/master/src/java.base/share/classes/java/util/ImmutableCollections.java)

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.