Skip to content

fix(inkless:switch): preserve classic prefix on partition increase during switch [KC-387] - #737

Draft
viktorsomogyi wants to merge 1 commit into
mainfrom
svv/KC-387-switch-dataloss
Draft

fix(inkless:switch): preserve classic prefix on partition increase during switch [KC-387]#737
viktorsomogyi wants to merge 1 commit into
mainfrom
svv/KC-387-switch-dataloss

Conversation

@viktorsomogyi

Copy link
Copy Markdown
Contributor

A CreatePartitions committed during a classic-to-diskless switch created the switching partition's control-plane logs row at high_watermark=0. That empty placeholder made the diskless LATEST report 0, so the consolidation fetcher truncated the classic prefix to 0 and the pre-switch records were lost.

  • createDisklessPartitions now creates only the newly added partitions [priorCount, count), so it never re-creates a switching partition's row.
  • init_diskless_log_v1 and the in-memory control plane now adopt the seal over an empty placeholder instead of no-oping, so the seal is authoritative whichever writer arrives first.

Adds a control-plane regression test on both backends and an end-to-end test that grows partitions mid-switch and verifies the classic prefix stays readable.

…ring switch [KC-387]

A CreatePartitions committed during a classic-to-diskless switch created the
switching partition's control-plane logs row at high_watermark=0. That empty
placeholder made the diskless LATEST report 0, so the consolidation fetcher
truncated the classic prefix to 0 and the pre-switch records were lost.

- createDisklessPartitions now creates only the newly added partitions
  [priorCount, count), so it never re-creates a switching partition's row.
- init_diskless_log_v1 and the in-memory control plane now adopt the seal over
  an empty placeholder instead of no-oping, so the seal is authoritative
  whichever writer arrives first.

Adds a control-plane regression test on both backends and an end-to-end test
that grows partitions mid-switch and verifies the classic prefix stays readable.

Co-authored-by: Cursor <cursoragent@cursor.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Prevents classic-prefix loss when partitions are added during a classic-to-diskless switch.

Changes:

  • Creates control-plane rows only for newly added partitions.
  • Makes the committed seal authoritative over empty placeholders.
  • Adds backend and end-to-end regression coverage; regenerates jOOQ schema metadata.

Reviewed changes

Copilot reviewed 7 out of 125 changed files in this pull request and generated 1 comment.

Show a summary per file
File Description
core/src/main/scala/kafka/server/ControllerApis.scala Passes the prior partition count to control-plane creation.
core/src/test/java/kafka/server/InklessConsolidatedDisklessTopicsTest.java Adds the end-to-end regression test.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/CreateTopicAndPartitionsRequest.java Adds the first-partition range bound.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/InMemoryControlPlane.java Adopts seals over empty placeholders.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/TopicsAndPartitionsCreateJob.java Inserts only the requested partition range.
storage/inkless/src/main/resources/db/migration/V23__Init_diskless_log_authoritative_seal.sql Adds guarded authoritative-seal upsert behavior.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/AbstractControlPlaneTest.java Tests seal adoption on both backends.
storage/inkless/src/main/jooq/org/jooq/generated/UDTs.java Regenerates schema-23 metadata.
storage/inkless/src/main/jooq/org/jooq/generated/DefaultSchema.java Regenerates schema-23 metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Domains.java Regenerates schema-23 metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Indexes.java Regenerates schema-23 metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Keys.java Regenerates schema-23 metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Routines.java Regenerates schema-23 metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Tables.java Regenerates schema-23 metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/* Regenerates all changed enum bindings for schema 23.
storage/inkless/src/main/jooq/org/jooq/generated/routines/* Regenerates all changed routine bindings for schema 23.
storage/inkless/src/main/jooq/org/jooq/generated/tables/* Regenerates all changed table bindings for schema 23.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/* Regenerates all changed table-record bindings for schema 23.
storage/inkless/src/main/jooq/org/jooq/generated/udt/* Regenerates all changed UDT bindings for schema 23.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/* Regenerates all changed UDT path bindings for schema 23.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/* Regenerates all changed UDT record bindings for schema 23.
Suppressed comments (1)

core/src/main/scala/kafka/server/ControllerApis.scala:1006

  • Add a focused controller test that captures the CreateTopicAndPartitionsRequest and verifies firstPartition is the pre-increase count. The new end-to-end test can still pass if this code sends 0, because the authoritative-seal upsert independently repairs that placeholder, and the existing control-plane test uses the three-argument constructor explicitly starting at 0. Thus the stated “only newly added partitions” behavior is currently unprotected.
              val firstPartition = priorPartitionCounts.getOrElse(req.name(), 0)
              Some(new CreateTopicAndPartitionsRequest(topicIdOrError.result(), req.name(), firstPartition, req.count()))

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Comment on lines +958 to +960
val priorPartitionCounts: Map[String, Int] = topics.asScala
.map(t => t.name() -> metadataCache.numPartitions(t.name()).orElse(0).intValue())
.toMap
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants