Added Applications Events for application-level metrics and custom label support in KafkaSink

[aashishchauhan06]

While the existing KafkaSink provides detailed stage, executor, and query metrics, there is currently no built-in support for:

• Application-level aggregated metrics
• Custom metadata labels via Spark configuration
• applications_started and applications_ended events

This PR addresses these gaps while preserving backward compatibility with the existing KafkaSink. This PR adds:

• Application-level metrics:
• Executor counts
• Stage counts
• Task counts
• Support for app lablels via Spark configuration: spark.sparkmeasure.appLabels.*
• Enhanced applications_started and applications_ended events enriched with metadata

Happy to adjust naming, packaging, or documentation if needed.

[LucaCanali]

Hi,
Thanks for the contribution, this is useful work. I think it should be introduced as a new sink version rather than changing KafkaSink behavior in place.

This PR adds substantial new semantics and payload fields (application lifecycle events, counters, labels, config payload), so it can break existing consumers that rely on the current Kafka message contract. Could you please move this into a new listener class such as KafkaSinkV2 ?

Also, before merge, I’d ask to fix configurations -> conf.getAll.toMap in applications_ended can leak secrets; please remove it or apply a strict allowlist/redaction strategy.

If you go with KafkaSinkV2, please also add documentation with:

• schema docs for new event types/fields,
• tests for emitted payload correctness (especially app start/end and counters),
• clear compatibility note for existing users of KafkaSink.

Thank you,
Luca

[aashishchauhan06]

Hi, Thanks for the contribution, this is useful work. I think it should be introduced as a new sink version rather than changing KafkaSink behavior in place.

This PR adds substantial new semantics and payload fields (application lifecycle events, counters, labels, config payload), so it can break existing consumers that rely on the current Kafka message contract. Could you please move this into a new listener class such as KafkaSinkV2 ?

Also, before merge, I’d ask to fix configurations -> conf.getAll.toMap in applications_ended can leak secrets; please remove it or apply a strict allowlist/redaction strategy.

If you go with KafkaSinkV2, please also add documentation with:

• schema docs for new event types/fields,
• tests for emitted payload correctness (especially app start/end and counters),
• clear compatibility note for existing users of KafkaSink.

Thank you, Luca

Thanks for the detailed review and for the suggestions — really appreciate you taking the time to go through the changes. I’ll push an updated version of the PR once these changes are in place.

[pp-achauhan]

Hi, Thanks for the contribution, this is useful work. I think it should be introduced as a new sink version rather than changing KafkaSink behavior in place.
This PR adds substantial new semantics and payload fields (application lifecycle events, counters, labels, config payload), so it can break existing consumers that rely on the current Kafka message contract. Could you please move this into a new listener class such as KafkaSinkV2 ?
Also, before merge, I’d ask to fix configurations -> conf.getAll.toMap in applications_ended can leak secrets; please remove it or apply a strict allowlist/redaction strategy.
If you go with KafkaSinkV2, please also add documentation with:

• schema docs for new event types/fields,
• tests for emitted payload correctness (especially app start/end and counters),
• clear compatibility note for existing users of KafkaSink.

Thank you, Luca

Thanks for the detailed review and for the suggestions — really appreciate you taking the time to go through the changes. I’ll push an updated version of the PR once these changes are in place.

@LucaCanali Please review PR, i have modified PR to have backward compatibility.

[LucaCanali]

Thank you @pp-achauhan !