Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
devinrsmith wants to merge 1 commit into
base: rc/v41.x
Choose a base branch
from
Open

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

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 73 additions & 4 deletions extensions/kafka/src/main/java/io/deephaven/kafka/KafkaTools.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,71 @@ protected abstract KeyOrValueIngestData getIngestData(
protected abstract KeyOrValueProcessor getProcessor(
TableDefinition tableDef,
KeyOrValueIngestData data);

private KeyOrValueIngestData getIngestDataAndIncrementColumnIndex(
final KeyOrValue keyOrValue,
final SchemaRegistryClient schemaRegistryClient,
final Map<String, ?> configs,
final MutableInt nextColumnIndexMut,
final 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. We are trying to be "lenient" for impls
// that improperly update nextColumnIndexMut, or setting the wrong index on simpleColumnIndex, and
// warn them that it appears they are doing something wrong. If an implementation is improperly adding
// to columnsDefinitionOut though, this is a more serious error.
final KeyOrValueIngestData ingestData = getIngestData(keyOrValue, schemaRegistryClient, configs,
nextColumnIndexMut, columnDefinitionsOut);
// Note: choosing to _not_ log warning when caller is mutating the index "incorrectly", but keeping code
// commented out here for future reference.
// final int addedIx = nextColumnIndexMut.get() - ixPre;
final int addedColumns = columnDefinitionsOut.size() - sizePre;
if (ingestData == null) {
// ignore case (or, possibly 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 columnDefinitionsOut without returning ingest data");
}
// if (addedIx != 0) {
// log.warn().append(
// "Ignore KeyOrValueSpec getIngestData is improperly mutating the index; manually correcting...")
// .endl();
// }
} else if (ingestData.isSimple()) {
// simple case:
// expecting exactly one increment and one column
if (addedColumns != 1) {
throw new IllegalStateException(
"Simple KeyOrValueSpec getIngestData is not adding exactly one column");
}
// if (addedIx != 1) {
// log.warn().append(
// "Simple KeyOrValueSpec did not properly mutate the index by 1; manually correcting...")
// .endl();
// }
// if (ingestData.simpleColumnIndex != ixPre) {
// log.warn().append("Simple KeyOrValueSpec set a bad simpleColumnIndex; manually correcting...")
// .endl();
// }
// The fact that we are setting this here means that this would ideally not be the responsibility of
// the implementation...
ingestData.simpleColumnIndex = ixPre;
} else {
// complex case:
// expecting no increments - we'll increase the index based on any columns that were added
// if (addedIx != 0) {
// log.warn().append(
// "Complex KeyOrValueSpec getIngestData is improperly mutating the index; manually correcting...")
// .endl();
// }
}
// The fact that we are setting this here means that this would ideally not be the responsibility of the
// implementation...
nextColumnIndexMut.set(ixPre + addedColumns);
return ingestData;
}
}

private static final KeyOrValueSpec FROM_PROPERTIES = new SimpleConsume(null, null);
Expand Down Expand Up @@ -1273,10 +1338,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.getIngestDataAndIncrementColumnIndex(
KeyOrValue.KEY, schemaRegistryClient, configs, nextColumnIndex, columnDefinitions);
final KeyOrValueIngestData valueIngestData = valueSpec.getIngestDataAndIncrementColumnIndex(
KeyOrValue.VALUE, schemaRegistryClient, configs, nextColumnIndex, columnDefinitions);

final TableDefinition tableDefinition = TableDefinition.of(columnDefinitions);
publisherParametersBuilder.setTableDefinition(tableDefinition);
Expand Down Expand Up @@ -1670,6 +1735,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 {
Expand Down
Loading
⚡