[CLI-3630] Support Schema IDs in headers

[Steven Gagniere (sgagniere)]

Release Notes

Breaking Changes

• PLACEHOLDER

New Features

• confluent kafka topic consume now supports consuming messages with schema IDs serialized in the header

Bug Fixes

• PLACEHOLDER

Checklist

• I have successfully built and used a custom CLI binary, without linter issues from this PR.
• I have clearly specified in the What section below whether this PR applies to Confluent Cloud, Confluent Platform, or both.
• I have verified this PR in Confluent Cloud pre-prod or production environment, if applicable.
• I have verified this PR in Confluent Platform on-premises environment, if applicable.
• I have attached manual CLI verification results or screenshots in the Test & Review section below.
• I have added appropriate CLI integration or unit tests for any new or updated commands and functionality.
• I confirm that this PR introduces no breaking changes or backward compatibility issues.
• I have indicated the potential customer impact if something goes wrong in the Blast Radius section below.
• I have put checkmarks below confirming that the feature associated with this PR is enabled in:
  • Confluent Cloud prod
  • Confluent Cloud stag or devel
  • Confluent Platform
  • Check this box if the feature is enabled for certain organizations only

What

Support schema IDs in headers for confluent kafka topic [produce | consume] for both Cloud and On-prem.

Blast Radius

The update to the functionality is additive, but the implementation required changes to the serdes package. So in the worst case bugs could impact the deserialization of json/avro/protobuf messages in kafka topic consume and asyncapi export.

References

Test & Review

Manual testing: https://docs.google.com/document/d/13e62TwsqCSwT6FU587Gm03Vz_VNFP4c9XrCw2anK8oI/edit?usp=sharing

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Copilot]

Pull Request Overview

This PR updates the confluent-kafka-go library to version 2.11.0 to support serializing and deserializing schema IDs in message headers instead of the traditional message prefix. The changes introduce a new --schema-id-header flag for the confluent kafka topic produce command and automatically support header-based schema IDs in the consume command.

Key changes:

• Updated confluent-kafka-go dependency from v2.8.0 to v2.11.0
• Added --schema-id-header flag to serialize schema IDs in headers for produce operations
• Refactored serialization/deserialization interfaces to support headers and updated all provider implementations
• Consolidated Schema Registry client initialization logic and improved test coverage for header-based schema IDs

Reviewed Changes

Copilot reviewed 23 out of 24 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
go.mod	Updated confluent-kafka-go dependency to v2.11.0
pkg/serdes/serdes.go	Added header support to serialization/deserialization interfaces and consolidated SR client initialization
pkg/serdes/*_serialization_provider.go	Updated all serialization providers to return headers alongside serialized data
pkg/serdes/*_deserialization_provider.go	Updated all deserialization providers to accept headers and use shared SR client initialization
pkg/serdes/serdes_test.go	Added comprehensive test coverage for header-based schema ID serialization/deserialization
internal/kafka/command_topic_produce.go	Added --schema-id-header flag support and updated message serialization to include headers
internal/kafka/confluent_kafka.go	Refactored consume logic to support header-based schema IDs and removed legacy RequestSchema method
internal/kafka/confluent_kafka_configs.go	Enhanced debug logging configuration for producer/consumer
internal/asyncapi/command_export.go	Updated to use new deserialization interface with headers
test/fixtures/output/kafka/topic/*.golden	Updated test fixtures for help text and error messages

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

[nitpick] Removed blank line between the deserializer creation and error check. While this works, the previous formatting with the blank line was more consistent with the style used elsewhere in the codebase.

Suggested change

	deser, err := avrov2.NewDeserializer(serdeClient, serdeType, serdeConfig)
	deser, err := avrov2.NewDeserializer(serdeClient, serdeType, serdeConfig)

[sonarqube-confluent]

• 53.80% Coverage on New Code (is less than 80.00%)

Analysis Details

8 Issues

• 0 Bugs
• 0 Vulnerabilities
• 8 Code Smells

Coverage and Duplications

• 53.80% Coverage (78.30% Estimated after merge)
• No duplication information (0.20% Estimated after merge)

Project ID: cli

View in SonarQube

[sonarqube-confluent]

Quality Gate failed

Failed conditions
57.5% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

[Channing Dong (channingdong)]

As a precautionary sanity check, can we add one more manual test case that produces and consumes a plain string message (no schema involved)?

[sonarqube-confluent]

Quality Gate failed

Failed conditions
57.8% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

[Channing Dong (channingdong)]

LGTM