This is an automated email from the ASF dual-hosted git repository.
soumyakanti3578 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new c23e5b54fe1 HIVE-29909: Kafka handler should reject class-loading
client properties set as table properties (#6870)
c23e5b54fe1 is described below
commit c23e5b54fe166fccdc9e4ad452662d70aa326877
Author: Ruben Quesada Lopez <[email protected]>
AuthorDate: Tue Oct 13 16:37:38 2026 +0100
HIVE-29909: Kafka handler should reject class-loading client properties set
as table properties (#6870)
---
.../hadoop/hive/kafka/KafkaStorageHandler.java | 16 +-
.../org/apache/hadoop/hive/kafka/KafkaUtils.java | 175 ++++++++++-
.../hadoop/hive/kafka/KafkaStorageHandlerTest.java | 86 ++++++
.../apache/hadoop/hive/kafka/KafkaUtilsTest.java | 327 +++++++++++++++++++++
4 files changed, 595 insertions(+), 9 deletions(-)
diff --git
a/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java
b/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java
index 8f4b22dabd3..aa8b71ffa96 100644
---
a/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java
+++
b/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java
@@ -29,6 +29,7 @@
import org.apache.hadoop.hive.metastore.DefaultHiveMetaHook;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
import org.apache.hadoop.hive.metastore.TableType;
+import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
import org.apache.hadoop.hive.metastore.api.LockType;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Table;
@@ -104,6 +105,7 @@
}
private void configureCommonProperties(TableDesc tableDesc, Map<String,
String> jobProperties) {
+ KafkaUtils.checkTableDescProperties(tableDesc.getProperties());
String topic =
tableDesc.getProperties().getProperty(KafkaTableProperties.HIVE_KAFKA_TOPIC.getName(),
"");
if (topic.isEmpty()) {
throw new IllegalArgumentException("Kafka topic missing set table
property->"
@@ -191,9 +193,7 @@ private void configureCommonProperties(TableDesc tableDesc,
Map<String, String>
.startsWith(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX))
.forEach(entry -> {
String key =
entry.getKey().substring(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX.length() + 1);
- if (KafkaUtils.FORBIDDEN_PROPERTIES.contains(key)) {
- throw new IllegalArgumentException("Not suppose to set Kafka
Property " + key);
- }
+ KafkaUtils.checkTableClientProperty(key, entry.getValue());
properties.put(key, entry.getValue());
});
return new KafkaStorageHandlerInfo(topic, properties);
@@ -231,9 +231,7 @@ private Properties buildProducerProperties(Table table) {
.startsWith(KafkaUtils.PRODUCER_CONFIGURATION_PREFIX))
.forEach(entry -> {
String key =
entry.getKey().substring(KafkaUtils.PRODUCER_CONFIGURATION_PREFIX.length() + 1);
- if (KafkaUtils.FORBIDDEN_PROPERTIES.contains(key)) {
- throw new IllegalArgumentException("Not suppose to set Kafka
Property " + key);
- }
+ KafkaUtils.checkTableClientProperty(key, entry.getValue());
properties.put(key, entry.getValue());
});
return properties;
@@ -385,6 +383,7 @@ private String getQueryId() {
if (!table.getTableType().equals(TableType.EXTERNAL_TABLE.toString())) {
throw new MetaException(KAFKA_STORAGE_HANDLER + " supports only " +
TableType.EXTERNAL_TABLE);
}
+ KafkaUtils.checkTableProperties(table.getParameters());
Arrays.stream(KafkaTableProperties.values())
.filter(KafkaTableProperties::isMandatory)
.forEach(key ->
Preconditions.checkNotNull(table.getParameters().get(key.getName()),
@@ -401,6 +400,11 @@ private String getQueryId() {
}
+ @Override public void preAlterTable(Table table, EnvironmentContext context)
throws MetaException {
+ super.preAlterTable(table, context);
+ KafkaUtils.checkTableProperties(table.getParameters());
+ }
+
@Override public void commitCreateTable(Table table) throws MetaException {
commitInsertTable(table, false);
}
diff --git
a/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaUtils.java
b/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaUtils.java
index 9e74d49a187..371e2a6416c 100644
--- a/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaUtils.java
+++ b/kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaUtils.java
@@ -42,7 +42,10 @@
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SecurityConfig;
import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.InvalidTopicException;
@@ -50,6 +53,7 @@
import org.apache.kafka.common.errors.SecurityDisabledException;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.errors.UnknownServerException;
+import org.apache.kafka.common.security.JaasContext;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.slf4j.Logger;
@@ -61,6 +65,7 @@
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.Base64;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
@@ -110,6 +115,172 @@ private KafkaUtils() {
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG));
+ /**
+ * System property letting an operator enable the {@link
#CLASS_NAME_PROPERTIES} for values
+ * configured globally in hive-site.xml. It does not enable them as table
properties.
+ */
+ static final String ALLOW_CLASS_NAME_PROPERTIES =
"hive.kafka.client.class.name.properties.allowed";
+
+ /**
+ * Kafka client properties whose value names one or more classes that the
client loads and
+ * instantiates in the JVM running the client. Deciding which classes the
Hive service loads is a
+ * deployment concern, so these belong to the site configuration an operator
controls rather than
+ * to per table DDL, and they are not accepted as table properties.
+ *
+ * <p>Most hold a single class name, while {@code interceptor.classes},
{@code metric.reporters}
+ * and {@code partition.assignment.strategy} hold a comma separated list.
Kafka types the two
+ * groups differently, {@code ConfigDef.Type.CLASS} against {@code
ConfigDef.Type.LIST}, and loads
+ * the first group while parsing the configuration and the second while
building the client, but
+ * both end up in {@code Utils.loadClass}, so they are treated alike
here.</p>
+ *
+ * <p>An operator who needs one of them, say a custom OAUTHBEARER callback
handler, sets it under
+ * the {@link #CONSUMER_CONFIGURATION_PREFIX} / {@link
#PRODUCER_CONFIGURATION_PREFIX} prefix in
+ * hive-site.xml and opts in with {@code
-Dhive.kafka.client.class.name.properties.allowed=true}.</p>
+ */
+ static final Set<String> CLASS_NAME_PROPERTIES =
ImmutableSet.of(SaslConfigs.SASL_LOGIN_CLASS,
+ SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
+ SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS,
+ SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG,
+ SecurityConfig.SECURITY_PROVIDERS_CONFIG,
+ CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
+ ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
+ ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
+ ProducerConfig.PARTITIONER_CLASS_CONFIG);
+
+ /**
+ * {@code config.providers.<name>.class} names a class the same way the
+ * {@link #CLASS_NAME_PROPERTIES} do, but it carries a user chosen name in
the middle, so it is
+ * matched by prefix rather than by set membership. The same prefix also
covers the
+ * {@code config.providers.<name>.param.*} keys that configure such a class.
+ */
+ static final String CONFIG_PROVIDERS_PREFIX = "config.providers.";
+
+ /**
+ * Checks a property whose value is known to come from a table property, so
the
+ * {@link #ALLOW_CLASS_NAME_PROPERTIES} opt-in does not apply.
+ *
+ * @param key the Kafka client property name, given without the Hive prefix.
+ * @param value the property value as the user supplied it.
+ * @throws IllegalArgumentException when a user must not set the property to
that value.
+ */
+ static void checkTableClientProperty(String key, String value) {
+ checkClientProperty(key, value, true);
+ }
+
+ /**
+ * Checks a property read out of a {@link Configuration}, where a value set
by an operator in
+ * hive-site.xml is indistinguishable from a table property, so the
+ * {@link #ALLOW_CLASS_NAME_PROPERTIES} opt-in applies.
+ *
+ * @param key the Kafka client property name, given without the Hive prefix.
+ * @param value the property value, which Hive has escaped if it came from a
table property.
+ * @throws IllegalArgumentException when a user must not set the property to
that value.
+ */
+ static void checkConfiguredClientProperty(String key, String value) {
+ checkClientProperty(key, value, false);
+ }
+
+ /**
+ * The single policy for one Kafka client property, name given without the
Hive prefix. There is
+ * one policy and it is applied everywhere; the only thing that varies
between call sites is
+ * whether the value's provenance is known, which is what {@code
provenanceIsTable} records.
+ *
+ * @param key the Kafka client property name, e.g. {@code sasl.login.class}.
+ * @param value the property value.
+ * @param provenanceIsTable true when the value is known to come from a
table property. False
+ * only for a {@link Configuration}, in which Hive has already merged
hive-site.xml with the table
+ * properties ({@code Utilities.copyTableJobPropertiesToConf}) so the
two can no longer be told apart.
+ * Since an operator legitimately needs the {@link
#CLASS_NAME_PROPERTIES} from hive-site.xml,
+ * that ambiguous case is the only one where {@link
#ALLOW_CLASS_NAME_PROPERTIES} can re-enable them.
+ * @throws IllegalArgumentException when a user must not set the property to
that value.
+ */
+ private static void checkClientProperty(String key, String value, boolean
provenanceIsTable) {
+ if (FORBIDDEN_PROPERTIES.contains(key)) {
+ throw new IllegalArgumentException("Not suppose to set Kafka Property "
+ key);
+ }
+ if (CLASS_NAME_PROPERTIES.contains(key) ||
key.startsWith(CONFIG_PROVIDERS_PREFIX)) {
+ if (provenanceIsTable) {
+ throw new IllegalArgumentException("Kafka Property " + key + "
configures a class that the Hive service "
+ + "would load and run, therefore it can not be set as a table
property.");
+ }
+ if (!Boolean.getBoolean(ALLOW_CLASS_NAME_PROPERTIES)) {
+ throw new IllegalArgumentException("Kafka Property " + key + "
configures a class that the Hive service "
+ + "would load and run, therefore it can not be set here. An
operator can set -D"
+ + ALLOW_CLASS_NAME_PROPERTIES + "=true to allow values configured
in hive-site.xml.");
+ }
+ }
+ if (SaslConfigs.SASL_JAAS_CONFIG.equals(key)) {
+ checkJaasConfig(value);
+ }
+ }
+
+ /**
+ * Rejects a {@link SaslConfigs#SASL_JAAS_CONFIG} value Kafka would not
accept, reusing Kafka's own
+ * mechanisms. Kafka applies the same check when the client is built;
running it here fails the DDL
+ * rather than the query.
+ *
+ * @param jaasConfig the user supplied {@code sasl.jaas.config} value.
+ * @throws IllegalArgumentException when the value does not parse or names a
refused login module.
+ */
+ static void checkJaasConfig(String jaasConfig) {
+ try {
+
JaasContext.loadClientContext(Collections.singletonMap(SaslConfigs.SASL_JAAS_CONFIG,
new Password(jaasConfig)));
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException("Refused value for Kafka Property " +
SaslConfigs.SASL_JAAS_CONFIG + ": "
+ + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Checks the Kafka client properties a user supplied as table properties,
before the table is
+ * created or altered.
+ *
+ * @param tableProperties the table parameters as supplied by the user, may
be null.
+ * @throws IllegalArgumentException when a user must not set one of the
table properties.
+ */
+ static void checkTableProperties(Map<String, String> tableProperties) {
+ if (tableProperties == null) {
+ return;
+ }
+ for (Map.Entry<String, String> entry : tableProperties.entrySet()) {
+ checkTableProperty(entry.getKey(), entry.getValue());
+ }
+ }
+
+ /**
+ * Same as {@link #checkTableProperties(Map)} for the {@link Properties}
view of a table.
+ *
+ * @param tableProperties the table properties, may be null.
+ * @throws IllegalArgumentException when a user must not set one of the
table properties.
+ */
+ static void checkTableDescProperties(Properties tableProperties) {
+ if (tableProperties == null) {
+ return;
+ }
+ for (String name : tableProperties.stringPropertyNames()) {
+ checkTableProperty(name, tableProperties.getProperty(name));
+ }
+ }
+
+ /**
+ * Checks one table property, whose name might carry the {@value
#CONSUMER_CONFIGURATION_PREFIX}
+ * or {@value #PRODUCER_CONFIGURATION_PREFIX} prefix.
+ */
+ private static void checkTableProperty(String rawKey, String value) {
+ String key = rawKey;
+ for (String prefix : new String[] {CONSUMER_CONFIGURATION_PREFIX + ".",
PRODUCER_CONFIGURATION_PREFIX + "."}) {
+ if (key.startsWith(prefix)) {
+ key = key.substring(prefix.length());
+ break;
+ }
+ }
+ try {
+ checkTableClientProperty(key, value);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Refused table property " + rawKey +
": " + e.getMessage(), e);
+ }
+ }
+
/**
* @param configuration Job configs
*
@@ -231,9 +402,7 @@ private static Map<String, String>
extractExtraProperties(final Configuration co
final Map<String, String> kafkaProperties =
configuration.getValByRegex("^" + prefix + "\\..*");
for (Map.Entry<String, String> entry : kafkaProperties.entrySet()) {
String key = entry.getKey().substring(prefix.length() + 1);
- if (FORBIDDEN_PROPERTIES.contains(key)) {
- throw new IllegalArgumentException("Not suppose to set Kafka Property
" + key);
- }
+ checkConfiguredClientProperty(key, entry.getValue());
builder.put(key, entry.getValue());
}
return builder.build();
diff --git
a/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaStorageHandlerTest.java
b/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaStorageHandlerTest.java
index 7d6f889f97d..bef51ade03a 100644
---
a/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaStorageHandlerTest.java
+++
b/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaStorageHandlerTest.java
@@ -29,6 +29,8 @@
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
@@ -143,4 +145,88 @@ public class KafkaStorageHandlerTest {
Assert.assertEquals(String.valueOf(Integer.MAX_VALUE),
producerProperties.get(ProducerConfig.RETRIES_CONFIG));
Assert.assertEquals("true",
producerProperties.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG));
}
+
+ private static Table kafkaTable(String key, String value) {
+ Table table = new Table();
+ table.setTableType(TableType.EXTERNAL_TABLE.toString());
+ table.putToParameters(KafkaTableProperties.HIVE_KAFKA_TOPIC.getName(),
TEST_TOPIC);
+
table.putToParameters(KafkaTableProperties.HIVE_KAFKA_BOOTSTRAP_SERVERS.getName(),
LOCALHOST_9291);
+ table.putToParameters(key, value);
+ return table;
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
preCreateTableRefusesDisallowedLoginModule()
+ throws MetaException {
+ new KafkaStorageHandler().preCreateTable(kafkaTable(
+ KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." +
SaslConfigs.SASL_JAAS_CONFIG,
+ "com.sun.security.auth.module.JndiLoginModule required
user.provider.url=\"ldap://ldap.example.com:389/ou=users\";";));
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
preCreateTableRefusesCallbackHandlerClass()
+ throws MetaException {
+ new KafkaStorageHandler().preCreateTable(kafkaTable(
+ KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." +
SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
+ "com.example.CustomCallbackHandler"));
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
preCreateTableRefusesUnprefixedEngineFactory()
+ throws MetaException {
+ new KafkaStorageHandler().preCreateTable(
+ kafkaTable(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG,
"com.example.CustomEngineFactory"));
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
preAlterTableRefusesCallbackHandlerClass()
+ throws MetaException {
+ new KafkaStorageHandler().preAlterTable(kafkaTable(
+ KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." +
SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS,
+ "com.example.CustomCallbackHandler"), null);
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
preCreateTableRefusesProducerSideEngineFactory()
+ throws MetaException {
+ new KafkaStorageHandler().preCreateTable(kafkaTable(
+ KafkaUtils.PRODUCER_CONFIGURATION_PREFIX + "." +
SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG,
+ "com.example.CustomEngineFactory"));
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
preCreateTableRefusesConfigProviderClass()
+ throws MetaException {
+ new KafkaStorageHandler().preCreateTable(kafkaTable(
+ KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." +
KafkaUtils.CONFIG_PROVIDERS_PREFIX + "custom.class",
+ "com.example.CustomConfigProvider"));
+ }
+
+ @Test public void preCreateTableAcceptsOrdinaryConsumerProperty() throws
MetaException {
+ Table table = kafkaTable(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX +
".max.poll.records", "5000");
+ new KafkaStorageHandler().preCreateTable(table);
+ Assert.assertEquals("5000",
+ table.getParameters().get(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX +
".max.poll.records"));
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
storageHandlerInfoRefusesCallbackHandlerClass()
+ throws MetaException {
+ KafkaStorageHandler kafkaStorageHandler = new KafkaStorageHandler();
+ kafkaStorageHandler.setConf(new JobConf());
+ kafkaStorageHandler.getStorageHandlerInfo(kafkaTable(
+ KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." +
SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
+ "com.example.CustomCallbackHandler"));
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
storageHandlerInfoRefusesConfigProviderClass()
+ throws MetaException {
+ KafkaStorageHandler kafkaStorageHandler = new KafkaStorageHandler();
+ kafkaStorageHandler.setConf(new JobConf());
+ kafkaStorageHandler.getStorageHandlerInfo(kafkaTable(
+ KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." +
KafkaUtils.CONFIG_PROVIDERS_PREFIX + "custom.class",
+ "com.example.CustomConfigProvider"));
+ }
+
+ @Test(expected = IllegalArgumentException.class) public void
storageHandlerInfoRefusesDisallowedLoginModule()
+ throws MetaException {
+ KafkaStorageHandler kafkaStorageHandler = new KafkaStorageHandler();
+ kafkaStorageHandler.setConf(new JobConf());
+ kafkaStorageHandler.getStorageHandlerInfo(kafkaTable(
+ KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." +
SaslConfigs.SASL_JAAS_CONFIG,
+ "com.sun.security.auth.module.JndiLoginModule required
user.provider.url=\"ldap://ldap.example.com:389/ou=users\";";));
+ }
}
diff --git
a/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaUtilsTest.java
b/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaUtilsTest.java
index b8c43b96a47..7ba35777c51 100644
--- a/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaUtilsTest.java
+++ b/kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaUtilsTest.java
@@ -25,13 +25,24 @@
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Writable;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* Test for Utility class.
@@ -65,6 +76,322 @@ public KafkaUtilsTest() {
KafkaUtils.consumerProperties(configuration);
}
+ @After public void clearClassNamePropertyOptIn() {
+ System.clearProperty(KafkaUtils.ALLOW_CLASS_NAME_PROPERTIES);
+ }
+
+ private static Configuration consumerConfig(String key, String value) {
+ Configuration configuration = new Configuration();
+ configuration.set("kafka.bootstrap.servers", "localhost:9090");
+ configuration.set(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "." + key,
value);
+ return configuration;
+ }
+
+ private static final String CUSTOM_CLASS = "com.example.CustomClass";
+ private static final String CUSTOM_CONFIG_PROVIDER =
"com.example.CustomConfigProvider";
+
+ @Test public void canNotSetClassNamingPropertiesOnConsumer() {
+ for (String key : KafkaUtils.CLASS_NAME_PROPERTIES) {
+ try {
+ KafkaUtils.consumerProperties(consumerConfig(key, CUSTOM_CLASS));
+ Assert.fail("Expected " + key + " to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
e.getMessage().contains(key));
+ }
+ }
+ }
+
+ @Test public void canNotSetClassNamingPropertiesOnProducer() {
+ for (String key : KafkaUtils.CLASS_NAME_PROPERTIES) {
+ Configuration configuration = new Configuration();
+ configuration.set("kafka.bootstrap.servers", "localhost:9090");
+ configuration.set(KafkaTableProperties.WRITE_SEMANTIC_PROPERTY.getName(),
+ KafkaOutputFormat.WriteSemantic.AT_LEAST_ONCE.name());
+ configuration.set(KafkaUtils.PRODUCER_CONFIGURATION_PREFIX + "." + key,
CUSTOM_CLASS);
+ try {
+ KafkaUtils.producerProperties(configuration);
+ Assert.fail("Expected " + key + " to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
e.getMessage().contains(key));
+ }
+ }
+ }
+
+ @Test public void operatorCanOptInToClassNamingPropertiesGlobally() {
+ System.setProperty(KafkaUtils.ALLOW_CLASS_NAME_PROPERTIES, "true");
+ Properties properties =
+
KafkaUtils.consumerProperties(consumerConfig(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
"com.acme.OAuth"));
+ Assert.assertEquals("com.acme.OAuth",
properties.getProperty(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ }
+
+ @Test public void optInDoesNotReachTableProperties() {
+ System.setProperty(KafkaUtils.ALLOW_CLASS_NAME_PROPERTIES, "true");
+ assertTablePropertyRefused(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG,
"com.example.CustomEngineFactory");
+ }
+
+ private static void assertTablePropertyRefused(String key, String value) {
+ assertTablePropertyRefused(key, value, key);
+ }
+
+ private static void assertTablePropertyRefused(String key, String value,
String expectedInMessage) {
+ try {
+ KafkaUtils.checkTableProperties(Collections.singletonMap(key, value));
+ Assert.fail("Expected table property " + key + " to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
e.getMessage().contains(expectedInMessage));
+ }
+ }
+
+ /**
+ * The un-prefixed spelling reaches the Kafka client through
copySSLProperties, so it has to be
+ * refused at DDL time, where we still know the value came from a table
property.
+ */
+ @Test public void unprefixedClassNamingTablePropertyIsRefused() {
+ for (String key : KafkaUtils.CLASS_NAME_PROPERTIES) {
+ assertTablePropertyRefused(key, CUSTOM_CLASS);
+ }
+ }
+
+ @Test public void prefixedClassNamingTablePropertiesAreRefused() {
+ for (String prefix : new String[]
{KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + ".",
+ KafkaUtils.PRODUCER_CONFIGURATION_PREFIX + "."}) {
+ for (String key : KafkaUtils.CLASS_NAME_PROPERTIES) {
+ assertTablePropertyRefused(prefix + key, CUSTOM_CLASS);
+ }
+ }
+ }
+
+ /**
+ * The config provider keys carry a user chosen name in the middle, so they
are matched by prefix.
+ */
+ @Test public void configProviderClassTablePropertyIsRefused() {
+ assertTablePropertyRefused(KafkaUtils.CONFIG_PROVIDERS_PREFIX +
"custom.class", CUSTOM_CONFIG_PROVIDER);
+ }
+
+ @Test public void prefixedConfigProviderClassTablePropertyIsRefused() {
+ assertTablePropertyRefused(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "."
+ + KafkaUtils.CONFIG_PROVIDERS_PREFIX + "custom.class",
CUSTOM_CONFIG_PROVIDER,
+ KafkaUtils.CONFIG_PROVIDERS_PREFIX + "custom.class");
+ }
+
+ /**
+ * The parameters handed to such a class share the prefix, so they are
refused with it. On their
+ * own they are inert, Kafka only reads them once a provider class is
configured.
+ */
+ @Test public void configProviderParameterTablePropertyIsRefused() {
+ assertTablePropertyRefused(KafkaUtils.CONFIG_PROVIDERS_PREFIX +
"custom.param.allowed.paths", "/etc");
+ }
+
+ @Test public void configProviderClassIsRefusedOnConfigurationPath() {
+ try {
+
KafkaUtils.consumerProperties(consumerConfig(KafkaUtils.CONFIG_PROVIDERS_PREFIX
+ "custom.class",
+ CUSTOM_CONFIG_PROVIDER));
+ Assert.fail("Expected the config provider class to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
+ e.getMessage().contains(KafkaUtils.CONFIG_PROVIDERS_PREFIX +
"custom.class"));
+ }
+ }
+
+ @Test public void operatorCanOptInToConfigProvidersGlobally() {
+ System.setProperty(KafkaUtils.ALLOW_CLASS_NAME_PROPERTIES, "true");
+ Properties properties = KafkaUtils.consumerProperties(
+ consumerConfig(KafkaUtils.CONFIG_PROVIDERS_PREFIX + "custom.class",
CUSTOM_CONFIG_PROVIDER));
+ Assert.assertEquals(CUSTOM_CONFIG_PROVIDER,
+ properties.getProperty(KafkaUtils.CONFIG_PROVIDERS_PREFIX +
"custom.class"));
+ }
+
+ @Test public void optInDoesNotReachConfigProvidersOnTableClientPath() {
+ System.setProperty(KafkaUtils.ALLOW_CLASS_NAME_PROPERTIES, "true");
+ try {
+ KafkaUtils.checkTableClientProperty(KafkaUtils.CONFIG_PROVIDERS_PREFIX +
"custom.class",
+ CUSTOM_CONFIG_PROVIDER);
+ Assert.fail("Expected the config provider class to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
+ e.getMessage().contains(KafkaUtils.CONFIG_PROVIDERS_PREFIX +
"custom.class"));
+ }
+ }
+
+ /**
+ * Kafka marks the properties it resolves into a class with {@link
ConfigDef.Type#CLASS}. Pinning
+ * that set against the two lists in KafkaUtils means a Kafka upgrade that
adds one of them fails
+ * here, rather than quietly arriving as a table property nothing looks at.
It cannot cover the
+ * handful Kafka types as a list or a plain string, those stay enumerated by
hand.
+ */
+ @Test public void everyClassTypedKafkaPropertyIsListed() {
+ Set<String> classTyped = Stream.of(ConsumerConfig.configDef(),
ProducerConfig.configDef())
+ .flatMap(configDef -> configDef.configKeys().entrySet().stream())
+ .filter(entry -> entry.getValue().type == ConfigDef.Type.CLASS)
+ .map(Map.Entry::getKey)
+ .collect(Collectors.toSet());
+ // Guard against the lookup itself silently returning nothing, which would
leave the assertion
+ // below passing for the wrong reason.
+ Assert.assertTrue("Expected to find the class typed Kafka properties,
found " + classTyped,
+ classTyped.contains(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG)
+ && classTyped.contains(ProducerConfig.PARTITIONER_CLASS_CONFIG));
+
+ Set<String> listed = new HashSet<>(KafkaUtils.CLASS_NAME_PROPERTIES);
+ listed.addAll(KafkaUtils.FORBIDDEN_PROPERTIES);
+ List<String> unlisted = classTyped.stream().filter(name ->
!listed.contains(name)).sorted()
+ .collect(Collectors.toList());
+ Assert.assertEquals("Kafka properties resolved into a class that neither
list in KafkaUtils covers",
+ Collections.emptyList(), unlisted);
+ }
+
+ @Test public void ordinaryTablePropertiesAreAccepted() {
+ Map<String, String> tableProperties = new HashMap<>();
+ tableProperties.put(KafkaTableProperties.HIVE_KAFKA_TOPIC.getName(),
"a-topic");
+ tableProperties.put(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX +
".max.poll.records", "5000");
+ tableProperties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
"/etc/ssl/truststore.jks");
+ KafkaUtils.checkTableProperties(tableProperties);
+ }
+
+ @Test public void nullTablePropertiesAreAccepted() {
+ KafkaUtils.checkTableProperties(null);
+ }
+
+ @Test public void forbiddenPropertyIsRefusedOnTableClientPath() {
+ for (String key : KafkaUtils.FORBIDDEN_PROPERTIES) {
+ try {
+ KafkaUtils.checkTableClientProperty(key, "whatever");
+ Assert.fail("Expected " + key + " to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
e.getMessage().contains(key));
+ }
+ }
+ }
+
+ /**
+ * The operator opt-in covers values an operator put in hive-site.xml. A
value read back out of the
+ * table parameters is user supplied DDL whatever the opt-in says, so it
must stay refused.
+ */
+ @Test public void optInDoesNotReachTableClientPath() {
+ System.setProperty(KafkaUtils.ALLOW_CLASS_NAME_PROPERTIES, "true");
+ for (String key : KafkaUtils.CLASS_NAME_PROPERTIES) {
+ try {
+ KafkaUtils.checkTableClientProperty(key, CUSTOM_CLASS);
+ Assert.fail("Expected " + key + " to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
e.getMessage().contains(key));
+ }
+ }
+ }
+
+ @Test public void disallowedLoginModuleIsRefusedOnTableClientPath() {
+ try {
+ KafkaUtils.checkTableClientProperty(SaslConfigs.SASL_JAAS_CONFIG,
+ "com.sun.security.auth.module.JndiLoginModule required
user.provider.url=\"ldap://ldap.example.com:389/ou=users\";";);
+ Assert.fail("Expected the disallowed login module to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
e.getMessage().contains("JndiLoginModule"));
+ }
+ }
+
+ @Test public void ordinaryPropertyIsAcceptedOnTableClientPath() {
+ KafkaUtils.checkTableClientProperty("max.poll.records", "5000");
+ }
+
+ /**
+ * The functional guard rails now fail the DDL rather than only the query,
since checkTableProperties
+ * and the paths reading the parameters back share one gate.
+ */
+ @Test public void forbiddenTablePropertyIsRefusedAtDdl() {
+ assertTablePropertyRefused(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "."
+ + ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true",
ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
+ }
+
+ /**
+ * A login module Kafka itself refuses through {@code
org.apache.kafka.disallowed.login.modules} is
+ * reported at DDL time instead of at query time. The prefixed key also
covers the prefix
+ * stripping, which is why the message names the un-prefixed Kafka property.
+ */
+ @Test public void disallowedLoginModuleIsRefusedInJaasConfig() {
+ assertTablePropertyRefused(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "."
+ SaslConfigs.SASL_JAAS_CONFIG,
+ "com.sun.security.auth.module.JndiLoginModule required
user.provider.url=\"ldap://ldap.example.com:389/ou=users\";";,
+ "JndiLoginModule");
+ }
+
+ @Test public void secondDisallowedLoginModuleIsRefusedInJaasConfig() {
+ assertTablePropertyRefused(SaslConfigs.SASL_JAAS_CONFIG,
+ "com.sun.security.auth.module.LdapLoginModule required
userProvider=\"ldap://ldap.example.com:389/o=example\";";,
+ "LdapLoginModule");
+ }
+
+ @Test public void unparseableJaasConfigIsRefused() {
+ assertTablePropertyRefused(SaslConfigs.SASL_JAAS_CONFIG, "not a jaas
config at all");
+ }
+
+ @Test public void standardLoginModuleIsAcceptedInJaasConfig() {
+
KafkaUtils.checkJaasConfig("org.apache.kafka.common.security.plain.PlainLoginModule
required "
+ + "username=\"user\" password=\"pass\";");
+ }
+
+ /**
+ * A semicolon inside a quoted option value must not be mistaken for an
entry separator, which is
+ * why Kafka's own parser is used rather than splitting the value by hand.
+ */
+ @Test public void semicolonInsideQuotedValueIsAcceptedInJaasConfig() {
+
KafkaUtils.checkJaasConfig("org.apache.kafka.common.security.scram.ScramLoginModule
required "
+ + "username=\"user\" password=\"pa;ss\";");
+ }
+
+ /**
+ * Third party login modules such as the AWS MSK IAM one stay usable: the
policy is Kafka's deny
+ * list, not an allow list of the four modules Hive happens to know about.
+ */
+ @Test public void thirdPartyLoginModuleIsAcceptedInJaasConfig() {
+ KafkaUtils.checkJaasConfig("software.amazon.msk.auth.iam.IAMLoginModule
required;");
+ }
+
+ /**
+ * A value an operator put in hive-site.xml arrives literal and has to keep
working, which is the
+ * whole reason the Configuration path is more permissive than the table
property path.
+ */
+ @Test public void configuredJaasValueFromSiteConfigIsAccepted() {
+ Properties properties =
KafkaUtils.consumerProperties(consumerConfig(SaslConfigs.SASL_JAAS_CONFIG,
+ "org.apache.kafka.common.security.plain.PlainLoginModule required
username=\"user\" password=\"pass\";"));
+ Assert.assertNotNull(properties.getProperty(SaslConfigs.SASL_JAAS_CONFIG));
+ }
+
+ /**
+ * The value is checked on the Configuration path too. A table property
could only arrive here
+ * escapeJava'd by Utilities.copyTableJobPropertiesToConf, which Kafka's own
JAAS parser rejects
+ * anyway, so there is no working table this leniency would have protected.
+ */
+ @Test public void configuredJaasValueWithDisallowedLoginModuleIsRefused() {
+ try {
+
KafkaUtils.consumerProperties(consumerConfig(SaslConfigs.SASL_JAAS_CONFIG,
+ "com.sun.security.auth.module.JndiLoginModule required
user.provider.url=\"ldap://ldap.example.com:389/ou=users\";";));
+ Assert.fail("Expected the disallowed login module to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
e.getMessage().contains("JndiLoginModule"));
+ }
+ }
+
+ /**
+ * The compile time check runs on the table sourced TableDesc properties,
before Hive merges them
+ * into the job configuration, so a table property never becomes
indistinguishable from operator
+ * configuration. Without it the operator opt-in would also re-enable table
properties.
+ */
+ @Test public void tablePropertiesAreRefusedEvenWhenOperatorOptedIn() {
+ System.setProperty(KafkaUtils.ALLOW_CLASS_NAME_PROPERTIES, "true");
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX + "."
+ + SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
"com.example.CustomCallbackHandler");
+ try {
+ KafkaUtils.checkTableDescProperties(tableProperties);
+ Assert.fail("Expected the table property to be refused");
+ } catch (IllegalArgumentException e) {
+ Assert.assertTrue("Unexpected message: " + e.getMessage(),
+
e.getMessage().contains(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ }
+ }
+
+ @Test public void nullPropertiesViewIsAccepted() {
+ KafkaUtils.checkTableDescProperties(null);
+ }
+
@Test public void testSetupKafkaSslPropertiesNoSslIsUnchanged() {
Configuration config = new Configuration();
Properties props = new Properties();
