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

extensions/kafka/src/main/java/io/deephaven/kafka/KafkaTools.java

@@ -336,6 +336,45 @@ protected abstract KeyOrValueIngestData getIngestData(
             protected abstract KeyOrValueProcessor getProcessor(
                     TableDefinition tableDef,
                     KeyOrValueIngestData data);
+
+            private KeyOrValueIngestData ingestDataAndIncrement(
+                    KeyOrValue keyOrValue,
+                    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,
+                        nextColumnIndexMut, columnDefinitionsOut);
+                final int addedIx = nextColumnIndexMut.get() - ixPre;
+                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 (addedIx != 0 || addedColumns != 0) {
+                        throw new IllegalStateException(
+                                "KeyOrValueSpec getIngestData is modifying out-variables without returning ingest data");
+                    }
+                } else if (ingestData.isSimple()) {
+                    // simple case:
+                    // expecting exactly one increment and one column
+                    if (addedIx != 1 || addedColumns != 1) {
+                        throw new IllegalStateException(
+                                "Simple KeyOrValueSpec getIngestData is not adding exactly one index or one column");
+                    }
+                } else {
+                    // complex case:
+                    // expecting no increments - we'll increase the index based on any columns that were added
+                    if (addedIx != 0) {
+                        throw new IllegalStateException("Complex KeyOrValueSpec getIngestData is mutating the index");
+                    }
+                    nextColumnIndexMut.getAndAdd(addedColumns);
+                }
+                return ingestData;
+            }
         }
 
         private static final KeyOrValueSpec FROM_PROPERTIES = new SimpleConsume(null, null);
@@ -1273,10 +1312,10 @@ private static ConsumeStruct getConsumeStruct(
                     cc.setColumnIndex.setColumnIndex(publisherParametersBuilder, nextColumnIndex.getAndIncrement());
                 });
 
-        final KeyOrValueIngestData keyIngestData = keySpec.getIngestData(KeyOrValue.KEY,
-                schemaRegistryClient, configs, nextColumnIndex, columnDefinitions);
-        final KeyOrValueIngestData valueIngestData = valueSpec.getIngestData(KeyOrValue.VALUE,
-                schemaRegistryClient, configs, nextColumnIndex, columnDefinitions);
+        final KeyOrValueIngestData keyIngestData = keySpec.ingestDataAndIncrement(
+                KeyOrValue.KEY, schemaRegistryClient, configs, nextColumnIndex, columnDefinitions);
+        final KeyOrValueIngestData valueIngestData = valueSpec.ingestDataAndIncrement(
+                KeyOrValue.VALUE, schemaRegistryClient, configs, nextColumnIndex, columnDefinitions);
 
         final TableDefinition tableDefinition = TableDefinition.of(columnDefinitions);
         publisherParametersBuilder.setTableDefinition(tableDefinition);
@@ -1670,6 +1709,10 @@ public static class KeyOrValueIngestData {
         public int simpleColumnIndex = NULL_COLUMN_INDEX;
         public Function<Object, Object> toObjectChunkMapper = Function.identity();
         public Object extra;
+
+        private boolean isSimple() {
+            return simpleColumnIndex != NULL_COLUMN_INDEX;
+        }
     }
 
     private interface SetColumnIndex {