#13798 (comment)
When multiple kafka outputs have different partition configurations, they all use the same extension key kafkapartitionerextension.Type.String() (line 624), causing later outputs to overwrite earlier ones in extensionCfg. The elasticsearch branch avoids this by incorporating outputName into the extension ID via getBeatsAuthExtensionID(outputName). Consider using a unique extension ID per kafka output name, similar to how elasticsearch handles it.
- extensionCfg[kafkapartitionerextension.Type.String()] = partitioner
- } else {
- // Specifying empty map will make the extension use the default hash partitioner.
- extensionCfg[kafkapartitionerextension.Type.String()] = map[string]any{}
+ partitionerExtensionID := fmt.Sprintf("%s%s", OtelNamePrefix, outputName)
+ extensionCfg[partitionerExtensionID] = partitioner
+ exporterConfig["record_partitioner"] = map[string]any{
+ "extension": partitionerExtensionID,
+ }
+ } else {
+ // Specifying empty map will make the extension use the default hash partitioner.
+ partitionerExtensionID := fmt.Sprintf("%s%s", OtelNamePrefix, outputName)
+ extensionCfg[partitionerExtensionID] = map[string]any{}
+ exporterConfig["record_partitioner"] = map[string]any{
+ "extension": partitionerExtensionID,
+ }
#13798 (comment)
When multiple kafka outputs have different partition configurations, they all use the same extension key kafkapartitionerextension.Type.String() (line 624), causing later outputs to overwrite earlier ones in extensionCfg. The elasticsearch branch avoids this by incorporating outputName into the extension ID via getBeatsAuthExtensionID(outputName). Consider using a unique extension ID per kafka output name, similar to how elasticsearch handles it.