@@ -611,6 +611,163 @@ func innerStructFields(typ Type) []StructField {
611611 return nil
612612}
613613
614+ // miscEffectiveMaxVersion computes the effective max version for a non-top-level
615+ // DSL struct, since these don't have MaxVersion set. The effective max is the
616+ // highest of the flexible version and all field MinVersion values, recursing
617+ // into sub-structs.
618+ func miscEffectiveMaxVersion (s * Struct ) int {
619+ max := 0
620+ if s .FromFlexible && s .FlexibleAt > max {
621+ max = s .FlexibleAt
622+ }
623+ miscEffectiveMaxVersionFields (s .Fields , & max )
624+ return max
625+ }
626+
627+ func miscEffectiveMaxVersionFields (fields []StructField , max * int ) {
628+ for _ , f := range fields {
629+ if f .Tag >= 0 && f .MinVersion == - 1 {
630+ continue // tag-only fields don't define a version
631+ }
632+ if f .MinVersion > * max {
633+ * max = f .MinVersion
634+ }
635+ if inner := innerStructFields (f .Type ); inner != nil {
636+ miscEffectiveMaxVersionFields (inner , max )
637+ }
638+ }
639+ }
640+
641+ func TestValidateMiscDSLAgainstKafkaJSON (t * testing.T ) {
642+ kafkaDir := os .Getenv ("KAFKA_DIR" )
643+ if kafkaDir == "" {
644+ t .Skip ("KAFKA_DIR not set; skipping Kafka JSON validation" )
645+ }
646+
647+ initDSL (t )
648+
649+ // Build a map from name → DSL struct for non-top-level types.
650+ dslByName := make (map [string ]* Struct )
651+ for i := range newStructs {
652+ s := & newStructs [i ]
653+ if ! s .TopLevel {
654+ dslByName [s .Name ] = s
655+ }
656+ }
657+
658+ // Misc type mappings: Kafka JSON path (relative to KAFKA_DIR) → DSL name.
659+ type miscMapping struct {
660+ jsonPath string
661+ dslName string
662+ }
663+ mappings := []miscMapping {
664+ {"group-coordinator/src/main/resources/common/message/OffsetCommitKey.json" , "OffsetCommitKey" },
665+ {"group-coordinator/src/main/resources/common/message/OffsetCommitValue.json" , "OffsetCommitValue" },
666+ {"group-coordinator/src/main/resources/common/message/GroupMetadataKey.json" , "GroupMetadataKey" },
667+ {"group-coordinator/src/main/resources/common/message/GroupMetadataValue.json" , "GroupMetadataValue" },
668+ {"transaction-coordinator/src/main/resources/common/message/TransactionLogKey.json" , "TxnMetadataKey" },
669+ {"transaction-coordinator/src/main/resources/common/message/TransactionLogValue.json" , "TxnMetadataValue" },
670+ {"clients/src/main/resources/common/message/DefaultPrincipalData.json" , "DefaultPrincipalData" },
671+ {"clients/src/main/resources/common/message/EndTxnMarker.json" , "EndTxnMarker" },
672+ {"clients/src/main/resources/common/message/LeaderChangeMessage.json" , "LeaderChangeMessage" },
673+ }
674+
675+ for _ , m := range mappings {
676+ jsonPath := filepath .Join (kafkaDir , m .jsonPath )
677+ data , err := os .ReadFile (jsonPath ) //nolint:gosec // path is constructed from test constant + env var, not user input
678+ if err != nil {
679+ t .Errorf ("reading %s: %v" , m .jsonPath , err )
680+ continue
681+ }
682+ cleaned := stripJSONComments (data )
683+ var msg kafkaMessage
684+ if err := json .Unmarshal (cleaned , & msg ); err != nil {
685+ t .Errorf ("parsing %s: %v" , m .jsonPath , err )
686+ continue
687+ }
688+
689+ dsl , ok := dslByName [m .dslName ]
690+ if ! ok {
691+ t .Errorf ("%s: DSL struct %q not found" , m .jsonPath , m .dslName )
692+ continue
693+ }
694+
695+ t .Run (m .dslName , func (t * testing.T ) {
696+ validateMiscMessage (t , msg , dsl )
697+ })
698+ }
699+ }
700+
701+ func validateMiscMessage (t * testing.T , msg kafkaMessage , dsl * Struct ) {
702+ t .Helper ()
703+
704+ validVR := parseVersionRange (msg .ValidVersions )
705+ flexVR := parseVersionRange (msg .FlexibleVersions )
706+
707+ jsonMax := validVR .maxVer ()
708+ dslMax := miscEffectiveMaxVersion (dsl )
709+
710+ if jsonMax >= 0 && dslMax > jsonMax {
711+ t .Errorf ("max version: DSL %d > JSON %d" , dslMax , jsonMax )
712+ } else if jsonMax >= 0 && dslMax < jsonMax {
713+ fields := collectMissingFields (msg .Name , dslMax + 1 , jsonMax , msg .Fields )
714+ detail := fmt .Sprintf ("DSL v%d < JSON v%d" , dslMax , jsonMax )
715+ if len (fields ) > 0 {
716+ detail += ", new fields: " + strings .Join (fields , ", " )
717+ }
718+ t .Errorf ("max version: %s" , detail )
719+ }
720+
721+ // Validate flexible version.
722+ if flexVR .none {
723+ if dsl .FromFlexible {
724+ t .Errorf ("flexible version: DSL has flexible at %d but JSON has none" , dsl .FlexibleAt )
725+ }
726+ } else {
727+ if ! dsl .FromFlexible {
728+ t .Errorf ("flexible version: JSON has flexible at %d but DSL has none" , flexVR .min )
729+ } else if dsl .FlexibleAt != flexVR .min {
730+ t .Errorf ("flexible version: DSL %d != JSON %d" , dsl .FlexibleAt , flexVR .min )
731+ }
732+ }
733+
734+ // Build commonStructs lookup.
735+ commons := make (map [string ]kafkaStruct )
736+ for _ , cs := range msg .CommonStructs {
737+ commons [cs .Name ] = cs
738+ }
739+
740+ // The DSL's "with version field" adds Version as the first field,
741+ // but coordinator JSON schemas don't include it. Skip it when the
742+ // JSON has no corresponding Version field.
743+ dslFields := dsl .Fields
744+ if dsl .WithVersionField && len (dslFields ) > 0 {
745+ jsonHasVersion := false
746+ for _ , jf := range msg .Fields {
747+ if strings .EqualFold (jf .Name , "Version" ) {
748+ jsonHasVersion = true
749+ break
750+ }
751+ }
752+ if ! jsonHasVersion && strings .EqualFold (dslFields [0 ].FieldName , "Version" ) {
753+ dslFields = dslFields [1 :]
754+ }
755+ }
756+
757+ flexibleAt := - 1
758+ if dsl .FromFlexible {
759+ flexibleAt = dsl .FlexibleAt
760+ }
761+
762+ maxV := dslMax
763+ if jsonMax >= 0 && jsonMax < maxV {
764+ maxV = jsonMax
765+ }
766+ for v := validVR .min ; v <= maxV ; v ++ {
767+ compareFieldsAtVersion (t , msg .Name , v , flexibleAt , msg .Fields , dslFields , commons )
768+ }
769+ }
770+
614771// collectMissingFields returns a summary of JSON fields new in versions fromV..toV.
615772func collectMissingFields (path string , fromV , toV int , jsonFields []kafkaField ) []string {
616773 var out []string
0 commit comments