JdbcSinkConfig fails to initialize if run in Turkish locale (tr-TR)

[LukeWoodward]

The class io.confluent.connect.jdbc.sink.JdbcSinkConfig fails to perform static initialization if the JVM is running in the Turkish (tr-TR) locale. This causes all sorts of problems attempting to use a JDBC sink connector in the Turkish locale.

Fortunately this class contains a main method, which provides a straightforward way to reproduce the problem.

Expected behaviour

Running the main method of the JdbcSinkConfig class when using the Turkish locale prints out information on how to configure a JDBC sink connection. (This is what happens if you run it in an English locale.)

Actual behaviour

D:\SomeFilePath>java -cp .\target\classes;kafka-clients-6.0.0-ccs.jar;connect-api-6.0.0-ccs.jar;slf4j-api-1.7.36.jar -Duser.language=TR -Duser.country=tr io.confluent.connect.jdbc.sink.JdbcSinkConfig
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Exception in thread "main" java.lang.ExceptionInInitializerError
Caused by: org.apache.kafka.common.config.ConfigException: Invalid value insert for configuration insert.mode: Invalid enumerator
        at io.confluent.connect.jdbc.sink.JdbcSinkConfig$EnumValidator.ensureValid(JdbcSinkConfig.java:658)
        at org.apache.kafka.common.config.ConfigDef$ConfigKey.<init>(ConfigDef.java:1131)
        at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:152)
        at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:172)
        at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:211)
        at io.confluent.connect.jdbc.sink.JdbcSinkConfig.<clinit>(JdbcSinkConfig.java:362)

I don't have a copy of the kafka-connect-jdbc JAR, I'm running the above from a download of the source code from GitHub plus the minimum number of JARs to get it to run. I didn't bother setting up SLF4J properly, hence the noise from it. I'm on Windows, but I don't believe the OS makes a difference here.

The root cause of the problem is the validation of enum properties within the EnumValidator inner class. It uses String.toLowerCase(), without specifying a locale, on the names of enum constants. For the insert.mode configuration property, the valid values are those that are the names of the InsertMode enum (INSERT, UPSERT and UPDATE), plus these names converted to lower-case using String.toLowerCase(). In the Turkish locale, these are ınsert, upsert and update. The default value for insert.mode is insert, and in the Turkish locale, this isn't one of the valid values. Hence the validation logic that runs during the JdbcSinkConfig class's static initializer throws an exception, and causes this class to fail static initialization.

Additionally, the exception message (Invalid enumerator) isn't particularly helpful. If the list of valid values was included in the error message then the problem would become clear more quickly.

I believe this bug is the root cause of this StackOverflow question.