Support `Struct` fields in chanel and message name mapping

[jaley]

Currently it's only possible to refer to the Kafka topic and the record key in our message.name and channel configuration properties, by making use of #{topic} or #{key} in those patterns.

This is quite limiting in common use cases, because:

• Kafka topics can be very big. It'll often be the case that the Ably Connector is configured to only consume from a single topic. At which point, using the #{topic} substitution in either pattern is no more useful than a string constant.
• Keys are used by Kafka producers to determine the topic partition that the record is posted to. This places constraints on what the key will need to contain, and given that Kafka clusters are typically multi-purpose, it often won't be reasonable to put exactly the content required for a unique Ably channel name in each record key.

The channel and message name mapping feature is most useful when users can refer to fields within a record. This is possible when a schema registry and an appropriate data converter is configured and the incoming records will therefore have Kafka Connect Struct data. The Ably connector already supports this, as it's needed for automatic conversion of data to JSON before sending over a channel. If, however, the incoming data is unstructured, we can't do this.

We would need to require that any field referenced can reasonably be converted to a string (for use in either channel name or message name) and that in the case of channel name mappings, the resulting channel name adheres to Ably's channel naming restrictions.

[sync-by-unito[bot]]

➤ Automation for Jira commented:

The link to the corresponding Jira issue is https://ably.atlassian.net/browse/SDK-3689

[jaley]

Initial thoughts on how to implement this:

• We should keep the grammar consistent with existing Ably Firehose routingKey patterns
  • #{...} denotes a substitution
  • access fields with . notation
  • access map and maybe array elements with [ ... ] notation, if needed
• While in Firehose, users can reference channelName and message.*, the root level objects need to be different in Kafka (as we're moving data in the opposite direction) and it makes sense to use Kafka nomenclature:
  • Users at least need to refer to the outgoing SinkRecord itself
  • Records have both keys and values, and both can have a structured schema associated with them
  • We already support topic and key (without nesting), but this means the key is already supported if it happens to be a String, or a byte array that we can interpret as string data.
• Kafka Connect schemata consist of these types:
  • Scalar types are generally easy to interpolate into a string channel or message name. We should allow strings, all precisions of integer, and probably booleans (though they seem less useful). Continuous values (all precisions of float) shouldn't be accepted, as that'll typically be an error.
  • Maps may be useful, as there will often be semi-structured data within records. We could support the [ ... ] notation as in routingKey for this, but it could be added later if needed.
  • Arrays are likely more niche, but no harder to support than maps, as they're effectively just integer-keyed maps.
• Topic partitions are available in the SinkRecord and this might be useful for channel mappings in some use cases? It'd be easy enough to support something like {topic.name} and {topic.partition}

Recommendation:

• Define a Stringable scalar to be something we can safely format into a message or channel name, including:
  • STRING
  • INT of any precision, cast to string
  • BOOLEAN cast to string
  • BYTES interpreted as a UTF-8 string
  • This is carrying forward existing functionality, though I'm not convinced it's 100% useful. Java's default encoding is UTF-16, so not positive this is a sensible default assumption, but it works today so probably better not to break it.
• If no schema registry has been set and the configuration contains message.name or channel key with a structured interpolation string pattern, that's an error and we should stop the sink task immediately to flag this, rather than waiting for data to arrive that we can't process.
  • Only topic (including .name and .partition) and key (if Stringable) can be used in interpolation without a schema registry
• Maintain existing support for #{key} and #{topic}:
  • topic is an alias for #{topic.name} and is always a string
  • key, if used this way, must be Stringable
• Add support for #{topic.name} and #{topic.partition}
• Add support for #{key.*} and #{value.*} as root level objects to reference fields within records:
  • a.b is used to access fields and implies a must be a STRUCT
  • the leaf object in a field reference must always be Stringable
• Assert that resulting channel names always adhere to the channel name restrictions (there's already code to do this)

Later:

• Add support for maps and arrays
  • Use the [ ] notation for element access
  • Note that maps have a schema for both keys and values, which complicates things
• Perhaps support the #{value.myField | hash(10)} hashing functionality in routingKey ?

Limiting support to just STRUCT and scalars for now means we can get away with a simple string split on the . character and regexes. Map and array accessors and the hashing notation may be the point at which we need to embed a parser library, which is a bit more work.

[jaley]

Closed in #141