[STREAM-841] Perf improvements for kafka connector

[amdox-karl]

Hi, as requested, related to ticket TS022015413

[RomainAnselin]

Adding for context: running a profiler of the kafka connector on a table leveraging some UDT showed a large amount of time spent on StructToUDTCodec.externalToInternal with 30% of time spent analysing the metadata. Flame graph available below:
jfr-noex.html

Quoting the OP, the changes can be summarized as follow

Heatmap showed 30% of compute was being spent in mapAndQueueRecord, specifically mapping struct->UDT.
kafka-sink/sink/src/main/java/com/datastax/oss/kafka/sink/KafkaStruct.java

• added kafkaSchema() accessor for stable schema-key caching.

kafka-sink/sink/src/main/java/com/datastax/oss/kafka/sink/codecs/StructToUDTCodec.java

• precomputed UDT metadata in fields (size, field names/types/internal names),
• added schema-keyed plan cache (ConcurrentMap<Schema, StructPlan>),
• moved expensive schema/type/codec setup into createPlan(...),
• per-record path now just runs cached bindings

Will follow up with a CQL schema if available

[RomainAnselin]

Adding a CQL DDL example as well as an example of the UDT structure and tables leveraged

-- Structure summary:
--   udt_01:        125 fields (text, int, bigint, boolean, double)
--   table_01:       35 columns + composite PK (4 partition + 3 clustering) + secondary index on one column
--   table_02:       29 columns + composite PK (4 partition + 3 clustering)
--   table_03:        7 columns + composite PK (2 partition + 4 clustering)
--   table_04:        6 columns + composite PK (4 partition + 2 clustering)

udt_structure.txt

[sandeep-ctds]

Making the cache size configurable could be a future improvement.

[amdox-karl]

Hi @sandeep-ctds I note the integration test failures. I am trying to understand if they are related in any way with the PR content? It does not look like that to me, but please let me know if you need further changes from my side.

[sandeep-ctds]

Hi @sandeep-ctds I note the integration test failures. I am trying to understand if they are related in any way with the PR content? It does not look like that to me, but please let me know if you need further changes from my side.

They are not related. We are trying to mitigate them.

[sandeep-ctds]

As repository variables are not accessible to PRs from forked repositories, the PR checks are failing as expected.
We will raise an internal PR with the same changes and verify if the checks pass there.

[sandeep-ctds]

Closing this in favour of the internal PR with same changes.
#166