@@ -80,7 +80,8 @@ public void autoConfigure(PropertyResolver kafkaClusterProperties,
80
80
kafkaClusterProperties .getProperty ("schemaRegistrySsl.keystoreLocation" , String .class ).orElse (null ),
81
81
kafkaClusterProperties .getProperty ("schemaRegistrySsl.keystorePassword" , String .class ).orElse (null ),
82
82
kafkaClusterProperties .getProperty ("ssl.truststoreLocation" , String .class ).orElse (null ),
83
- kafkaClusterProperties .getProperty ("ssl.truststorePassword" , String .class ).orElse (null )
83
+ kafkaClusterProperties .getProperty ("ssl.truststorePassword" , String .class ).orElse (null ),
84
+ kafkaClusterProperties .getProperty ("ssl.verifySsl" , Boolean .class ).orElse (true )
84
85
),
85
86
kafkaClusterProperties .getProperty ("schemaRegistryKeySchemaNameTemplate" , String .class ).orElse ("%s-key" ),
86
87
kafkaClusterProperties .getProperty ("schemaRegistrySchemaNameTemplate" , String .class ).orElse ("%s-value" ),
@@ -106,7 +107,8 @@ public void configure(PropertyResolver serdeProperties,
106
107
serdeProperties .getProperty ("keystoreLocation" , String .class ).orElse (null ),
107
108
serdeProperties .getProperty ("keystorePassword" , String .class ).orElse (null ),
108
109
kafkaClusterProperties .getProperty ("ssl.truststoreLocation" , String .class ).orElse (null ),
109
- kafkaClusterProperties .getProperty ("ssl.truststorePassword" , String .class ).orElse (null )
110
+ kafkaClusterProperties .getProperty ("ssl.truststorePassword" , String .class ).orElse (null ),
111
+ kafkaClusterProperties .getProperty ("ssl.verifySsl" , Boolean .class ).orElse (true )
110
112
),
111
113
serdeProperties .getProperty ("keySchemaNameTemplate" , String .class ).orElse ("%s-key" ),
112
114
serdeProperties .getProperty ("schemaNameTemplate" , String .class ).orElse ("%s-value" ),
@@ -136,7 +138,8 @@ private static SchemaRegistryClient createSchemaRegistryClient(List<String> urls
136
138
@ Nullable String keyStoreLocation ,
137
139
@ Nullable String keyStorePassword ,
138
140
@ Nullable String trustStoreLocation ,
139
- @ Nullable String trustStorePassword ) {
141
+ @ Nullable String trustStorePassword ,
142
+ boolean verifySsl ) {
140
143
Map <String , String > configs = new HashMap <>();
141
144
if (username != null && password != null ) {
142
145
configs .put (BASIC_AUTH_CREDENTIALS_SOURCE , "USER_INFO" );
@@ -166,6 +169,11 @@ private static SchemaRegistryClient createSchemaRegistryClient(List<String> urls
166
169
keyStorePassword );
167
170
}
168
171
172
+ if (!verifySsl ) { // TODO block above
173
+ configs .put (SchemaRegistryClientConfig .CLIENT_NAMESPACE + SslConfigs .SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG ,
174
+ "" );
175
+ }
176
+
169
177
return new CachedSchemaRegistryClient (
170
178
urls ,
171
179
1_000 ,
0 commit comments