...

Aims to add a corresponding validator to a configuration where a validator is missing.
so that the program finds these incorrect configurations during initialization.

Public Interfaces

...

Public classes that involve changes

1. ConfigDef
   Add a new validator NonNullAndEmptyString  to the ConfigDef class.
   
   
2. SslClientAuth
   SslClientAuth provides a static method names()  to get a list of all enumeration constant lowercase names.
   
   
3. BrokerSecurityConfigs
   Provides a new validator SaslEnabledMechanismsValidator  for validating the sasl.enabled.mechanisms  parameter.
   
   
4. CompressionType
   CompressionType provides a static method names()  to get a list of all enumeration constant lowercase names
   
   

...

1. AdminClientConfig

   config	validator
   security.protocol	in(SecurityProtocol.names().toArray(newString[0]))

   
   
   

2. ConsumerConfig

   config	validator
   group.instance.id	newConfigDef.NonEmptyString()
   key.deserializer	newConfigDef.NonNullValidator()
   value.deserializer	newConfigDef.NonNullValidator()
   security.protocol	in(SecurityProtocol.names().toArray(newString[0]))

   
   
   

3. ProducerConfig

   config	validator
   key.serializer	newConfigDef.NonNullValidator()
   value.serializer	newConfigDef.NonNullValidator()
   security.protocol	in(SecurityProtocol.names().toArray(newString[0]))
   compression.type	in(CompressionType.names().toArray(newString[0]))

   
   
   

4. SaslConfigs

   config	validator
   sasl.mechanism	newConfigDef.NonNullAndEmptyString()

   
   
   

5. MirrorClientConfig

   config	validator
   security.protocol	in(SecurityProtocol.names().toArray(newString[0]))

   
   
   

6. MirrorConnectorConfig

   config	validator
   security.protocol	in(SecurityProtocol.names().toArray(newString[0]))

   
   
   

7. MirrorMakerConfig

   config	validator
   security.protocol	in(SecurityProtocol.names().toArray(newString[0]))

   
   
   

8. WorkerConfig

   config	validator
   ssl.client.auth	in(SslClientAuth.names().toArray(newString[0]))

   
   
   

9. DistributedConfig

   config	validator
   security.protocol	in(SecurityProtocol.names().toArray(newString[0]))

   
   
   

10. KafkaConfig

    config	validator
    sasl.mechanism.controller.protocol	newConfigDef.NonNullAndEmptyString()
    authorizer.class.name	new ConfigDef.NonNullValidator()
    security.inter.broker.protocol	in(SecurityProtocol.names().toArray(new Array[String](0)):_*)
    sasl.mechanism.inter.broker.protocol	newConfigDef.NonNullAndEmptyString()
    sasl.enabled.mechanisms	newBrokerSecurityConfigs.SaslEnabledMechanismsValidator()
    compression.type	in(BrokerCompressionCodec.brokerCompressionOptions:_*)

    
    
    

11. StreamsConfig

    config	validator
    security.protocol	in(SecurityProtocol.names().toArray(newString[0]))

    
    

Proposed Changes

1.ConfigDef.NonNullAndEmptyString

...

For all the configurations mentioned above, if the value set by the user is legal, the user will not perceive any changes.
If the value set by the user is invalid, the corresponding validator will throw a ConfigException  when the ConfigDef.parse(Map<?, ?> props)  method is executed. The corresponding Kafka program will fail to initialize.

...

I had hoped to add another validator for the following configuration: ValidList.in("GSSAPI", "PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512", "OAUTHBEARER") .

sasl.mechanism
sasl.enabled.mechanisms
sasl.mechanism.controller.protocol
sasl.mechanism.inter.broker.protocol

...