Support extract-key kafka message transform

[akrambek]

Description

Support extract-key kafka message transform.

Fixes #1176

[jfallows]

This should be consistent with the others, either subject everywhere or message.payload.name everywhere.

The value of subject is passed as part of the URL to access schema registry.

/subjects/{0}/versions/{1}

So this should probably be based on the channel address topic with -value suffix, rather than relying on any naming convention alignment between message payload name and channel address (kafka topic).

Note: given that channels (kafka topics) can have more than one message type in the general case, seems like we actually want to potentially use topic-record name strategy instead of topic name strategy to differentiate the schemas for different channel (topic) messages.

In that case, it would seem to make sense to base the record part of the topic-record name strategy on the message name within the channel messages.

[jfallows]

Suggested change

	KafkaTopicTransformsConfigBuilder<KafkaTopicTransformsConfig> topicBuilder = KafkaTopicTransformsConfig.builder();
	String extractKey = object.containsKey(EXTRACT_KEY_NAME)
	? object.getString(EXTRACT_KEY_NAME)
	: null;
	topicBuilder.extractKey(extractKey);
	JsonObject headers = object.containsKey(EXTRACT_HEADERS_NAME) ? object.getJsonObject(EXTRACT_HEADERS_NAME) : null;
	if (headers != null)
	{
	for (Map.Entry<String, JsonValue> entry : headers.entrySet())
	{
	JsonString jsonString = (JsonString) entry.getValue();
	topicBuilder.extractHeader(entry.getKey(), jsonString.getString());
	}
	}
	return topicBuilder.build();
	KafkaTopicTransformsConfigBuilder<KafkaTopicTransformsConfig> transforms = KafkaTopicTransformsConfig.builder();
	if (object.containsKey(EXTRACT_KEY_NAME))
	{
	String extractKey = object.getString(EXTRACT_KEY_NAME);
	transforms.extractKey(extractKey);
	}
	if (object.containsKey(EXTRACT_HEADERS_NAME))
	{
	JsonObject headers = object.getJsonObject(EXTRACT_HEADERS_NAME);
	for (Map.Entry<String, JsonValue> entry : headers.entrySet())
	{
	String headerName = entry.getKey();
	JsonString headerValue = (JsonString) entry.getValue();
	transforms.extractHeader(headerName, headerValue.getString());
	}
	}
	return transforms.build();

[jfallows]

Include extract-key here please.

[jfallows]

Include extract-key here please.

[jfallows]

Make sure you have caught up to latest develop, I think this change is already present there.

[jfallows]

Just curious, what prompted this change?

[akrambek]

The script we are using has 4 bytes key1 and the TestConverter was returning 12345 which has 5 bytes and it was causing testware issue. I just changed TestConverter to return 1234 to match number of bytes in server script.

[jfallows]

Suggested change

	private final ModelConfigAdapter converter = new ModelConfigAdapter();
	private final KafkaTopicTransformsConfigAdapter transformsConverter = new KafkaTopicTransformsConfigAdapter();
	private final ModelConfigAdapter model = new ModelConfigAdapter();
	private final KafkaTopicTransformsConfigAdapter transforms = new KafkaTopicTransformsConfigAdapter();

These are not converters (means something else for us too) plus probably no need to include the type in the name.

[jfallows]

I think this is assuming we can specify either extract-key or extract-headers only once in the transforms array, right?

[jfallows]

I think this is assuming that extract-key is always processed after extract-key, which might be fine for now if we cannot extract the key from a header yet, but probably not sufficient for the general case later where these transforms are expected to execute in the order defined, the output of the previous feeding into the input of the next.

[jfallows]

Suggested change

	private AsyncapiChannelResolver channels;
	private AsyncapiMessageResolver messages;
	private final AsyncapiChannelResolver channels;
	private final AsyncapiMessageResolver messages;