Unhardwire kafka key in notifications

[fgalan]

Is your feature request related to a problem / use case? Please describe.

Kafka notificationa currently are using a hardwired key (equal to subscription id), which is not convenient in some case (see telefonicaid/kafnus#200 (comment))

Describe the solution you'd like

Add a new optional parameter in kafka/kafkaCustom https://github.com/telefonicaid/fiware-orion/blob/master/doc/manuals/orion-api.md#subscriptionnotificationkafkacustom for the key (key).

The value of that parameter would be a JEXL expression (note that a constant value, such null or a string is a particular case of valida JEXL expression).

Eg:

...
"notification": {
  "kafka": {
    "url": "kafka://broker1:9092,broker2:9092,broker3:9092",
    "topic": "sub1",
    "key": null
  }
}
...

...
"notification": {
  "kafka": {
    "url": "kafka://broker1:9092,broker2:9092,broker3:9092",
    "topic": "sub1",
    "key": "${id}"
  }
}
...

Describe alternatives you've considered
None

Describe why you need this feature

• To improve or simplify an scenario.

Do you have the intention to implement the solution

• Yes, I have the knowledge to implement this new feature.
• Yes, but I will need help.
• No, I do not have the skills.

[GregorioBlazquez]

This appears to be the point where sub id is used as a key to Kafka when producing messages:

fiware-orion/src/lib/kafka/KafkaConnectionManager.cpp

Lines 486 to 493 in 02abad9

	int resultCode = rd_kafka_producev(
	producer,
	RD_KAFKA_V_TOPIC(topic.c_str()),
	RD_KAFKA_V_KEY((void*)subscriptionId.data(), subscriptionId.size()),
	RD_KAFKA_V_VALUE(const_cast<char*>(content.data()), content.size()),
	RD_KAFKA_V_HEADERS(headers),
	RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
	RD_KAFKA_V_END);

[orianar]

I agree with the issue proposal to add an optional notification.kafka.key field (JEXL) to control the Kafka key. I would keep subscriptionId as the default for backward compatibility when key is not provided, and allow key: null (or the expression evaluating to null) to produce with a NULL key so messages can be better distributed across partitions. Otherwise, the evaluated value would be used as the key (e.g. ${id} or ${service}:${servicePath}:${id}).

As a minimal context to evaluate the expression, would exposing id, type, service, and servicePath be enough?

Additionally, I suggest keeping subscriptionId as the default and documenting an entity-based key as the recommended option, e.g. key: "${service}:${servicePath}:${type}:${id}".

[fgalan]

I would keep subscriptionId as the default for backward compatibility when key is not provided

Ok

As a minimal context to evaluate the expression, would exposing id, type, service, and servicePath be enough?

I wonder why not to include full context (i.e. those ones but also al attributes, metadata, etc.). Would it be more costly?

Additionally, I suggest keeping subscriptionId as the default and documenting an entity-based key as the recommended option, e.g. key: "${service}:${servicePath}:${type}:${id}"

Ok

[fgalan]

key or partitionKey for the new field? (later seems to be more precise...)