fix: DH-22265: increment Kafka column index for complex spec

[devinrsmith]

This fixes a bug where the column indexing is incorrectly calculated; it only manifests when the key spec is "complex" and the value spec is "simple".

Fixes #7884

[github-actions]

No docs changes detected for 5dc6e69

[Copilot]

Pull request overview

Fixes Kafka consume column-index calculation when combining a complex key spec with a simple value spec, addressing the “Chunk sizes don’t match” ingestion error reported in #7884.

Changes:

• Add KeyOrValueSpec.ingestDataAndIncrement(...) to adjust nextColumnIndex after complex ingest specs.
• Switch getConsumeStruct to use the new helper for both key and value ingest-data creation.
• Add a small helper (KeyOrValueIngestData.isSimple()) to distinguish simple vs non-simple ingest specs.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

This change fixes subtle column-indexing behavior that only shows up with a complex key spec + simple value spec (per DH-22265 / #7884), but there doesn’t appear to be regression coverage exercising getConsumeStruct/consumeToTable with that spec combination. Adding a unit/integration test that builds a TableDefinition via getTableDefinition (or runs a minimal ingest path) for complex-key + simple-value and asserts the resulting indices/column ordering would help prevent future regressions.

[devinrsmith]

Testing will be in a follow-up PR.

[rcaudy]

I think we're being a bit brittle.

[rcaudy]

Maybe:

            protected KeyOrValueIngestData getIngestData(
                    KeyOrValue keyOrValue,
                    SchemaRegistryClient schemaRegistryClient,
                    Map<String, ?> configs,
                    int nextColumnIndex,
                    List<ColumnDefinition<?>> columnDefinitionsOut) {
                return getIngestData(keyOrValue, schemaRegistryClient, configs, new MutableInt(nextColumnIndex),
                        columnDefinitionsOut);
            }

            private KeyOrValueIngestData getIngestDataAndIncrementColumnIndex(
                    final KeyOrValue keyOrValue,
                    final SchemaRegistryClient schemaRegistryClient,
                    Map<String, ?> configs,
                    MutableInt nextColumnIndexMut,
                    List<ColumnDefinition<?>> columnDefinitionsOut) {
                final int ixPre = nextColumnIndexMut.get();
                final int sizePre = columnDefinitionsOut.size();
                // It's unfortunate we have to do this; but the getIngestData signature and design of KeyOrValueSpec
                // would need a breaking change to improve in these regards.
                final KeyOrValueIngestData ingestData = getIngestData(keyOrValue, schemaRegistryClient, configs,
                        ixPre, columnDefinitionsOut);
                final int addedColumns = columnDefinitionsOut.size() - sizePre;
                if (ingestData == null) {
                    // ignore case (or, possible a complex impl that chooses to return null when it's empty):
                    // expecting no modifications
                    if (addedColumns != 0) {
                        throw new IllegalStateException(
                                "KeyOrValueSpec getIngestData is modifying out-variables without returning ingest data");
                    }
                } else {
                    nextColumnIndexMut.set(ixPre + addedColumns);
                }
                return ingestData;
            }
        }

[rcaudy]

Approving, but I liked my simpler version that absolved the implementations of any responsibility for mutating the index quite a lot more. This version has an awful lot of validation and warning trace that only benefits anyone if the current nonsense implementation is desirable long-term.

[rcaudy]

I think the commented-out warnings are ugly, but I'm willing to compromise here.