Skip to content

KafkaStore SSL support. #319

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Alex L (alexlod) merged 1 commit into from
May 12, 2016
Merged

KafkaStore SSL support. #319

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion bin/schema-registry-run-class
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ fi

# Memory options
if [ -z "$SCHEMA_REGISTRY_HEAP_OPTS" ]; then
SCHEMA_REGISTRY_HEAP_OPTS="-Xmx256M"
SCHEMA_REGISTRY_HEAP_OPTS="-Xmx512M"

@ewencp Ewen Cheslack-Postava (ewencp) Apr 28, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems quite large for what the schema registry is doing. Did you actually run into limits with the existing setting?

@alexlod Alex L (alexlod) Apr 29, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ewen Cheslack-Postava (@ewencp), yeah I was getting OOMEs. I didn't test whether a smaller value would work or not. Want me to test a smaller value, like 384?

fi

# JVM performance options
Expand Down
13 changes: 13 additions & 0 deletions core/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,19 @@
<classifier>test</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>

@ewencp Ewen Cheslack-Postava (ewencp) Apr 28, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Doesn't this already get pulled in via the kafka_${kafka.scala.version} dependency?

@alexlod Alex L (alexlod) Apr 29, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, tests won't compile without this dependency. It's needed for the use of TestSslUtils.

@ewencp Ewen Cheslack-Postava (ewencp) Apr 29, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ack, was misreading this as the normal, not test, jar.

<artifactId>kafka-clients</artifactId>
<version>${kafka.version}</version>
<classifier>test</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcpkix-jdk15on</artifactId>
<version>${bouncycastle.bcpkix.version}</version>

@ijuma Ismael Juma (ijuma) May 6, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are using bouncycastle outside of tests?

@alexlod Alex L (alexlod) May 9, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't believe so. I'll make the scope test and test on my local machine. Good catch!

<scope>test</scope>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
Expand Down
117 changes: 114 additions & 3 deletions core/src/main/java/io/confluent/kafka/schemaregistry/rest/SchemaRegistryConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@ public class SchemaRegistryConfig extends RestConfig {

private static final int SCHEMAREGISTRY_PORT_DEFAULT = 8081;

public static final String KAFKASTORE_SECURITY_PROTOCOL_SSL = "SSL";
public static final String KAFKASTORE_SECURITY_PROTOCOL_PLAINTEXT = "PLAINTEXT";

public static final String KAFKASTORE_CONNECTION_URL_CONFIG = "kafkastore.connection.url";
/**
* <code>kafkastore.zk.session.timeout.ms</code>
Expand Down Expand Up @@ -76,6 +79,36 @@ public class SchemaRegistryConfig extends RestConfig {
* <code>avro.compatibility.level</code>
*/
public static final String COMPATIBILITY_CONFIG = "avro.compatibility.level";
public static final String KAFKASTORE_SECURITY_PROTOCOL_CONFIG =

@ewencp Ewen Cheslack-Postava (ewencp) Apr 28, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we leverage the client security config support in Kafka to simplify a bunch of this (see ConfigDef.withClientSslSupport/ConfigDef.withClientSaslSupport)? Not sure it's possible given the additional prefix, but it seems easier to maintain and less error prone if we can just get the configs directly from Kafka instead of copy/pasting here.

@alexlod Alex L (alexlod) Apr 29, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I didn't realize these existed -- great! I agree that this may be a better route. I'll experiment with prepending kafkastore. to each.

@alexlod Alex L (alexlod) Apr 29, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change would require we use Kafka's ConfigDef, which we decided we wouldn't do now. So although it's a great idea, for now we won't do it.

@hachikuji Jason Gustafson (hachikuji) Apr 29, 2016

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe add a TODO if it's something we think we'll do eventually?

@Ishiihara Liquan Pei (Ishiihara) Apr 29, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Alex L (@alexlod) Any particular reason we are not using Kafka's ConfigDef?

@Ishiihara Liquan Pei (Ishiihara) Apr 29, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The reason it because we don't have defineOverride, correct?

@alexlod Alex L (alexlod) May 2, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll experiment with Liquan Pei (@Ishiihara)'s suggestion below, to change ConfigDef to support null defaults. Regardless, I'll add a TODO per Jason Gustafson (@hachikuji)'s suggestion.

"kafkastore.security.protocol";
public static final String KAFKASTORE_SSL_TRUSTSTORE_LOCATION_CONFIG =
"kafkastore.ssl.truststore.location";
public static final String KAFKASTORE_SSL_TRUSTSTORE_PASSWORD_CONFIG =
"kafkastore.ssl.truststore.password";
public static final String KAFKASTORE_SSL_KEYSTORE_LOCATION_CONFIG =
"kafkastore.ssl.keystore.location";
public static final String KAFKASTORE_SSL_TRUSTSTORE_TYPE_CONFIG =
"kafkastore.ssl.truststore.type";
public static final String KAFKASTORE_SSL_TRUSTMANAGER_ALGORITHM_CONFIG =
"kafkastore.ssl.trustmanager.algorithm";
public static final String KAFKASTORE_SSL_KEYSTORE_PASSWORD_CONFIG =
"kafkastore.ssl.keystore.password";
public static final String KAFKASTORE_SSL_KEYSTORE_TYPE_CONFIG =
"kafkastore.ssl.keystore.type";
public static final String KAFKASTORE_SSL_KEYMANAGER_ALGORITHM_CONFIG =
"kafkastore.ssl.keymanager.algorithm";
public static final String KAFKASTORE_SSL_KEY_PASSWORD_CONFIG =
"kafkastore.ssl.key.password";
public static final String KAFKASTORE_SSL_ENABLED_PROTOCOLS_CONFIG =
"kafkastore.ssl.enabled.protocols";
public static final String KAFKASTORE_SSL_PROTOCOL_CONFIG =
"kafkastore.ssl.protocol";
public static final String KAFKASTORE_SSL_PROVIDER_CONFIG =
"kafkastore.ssl.provider";
public static final String KAFKASTORE_SSL_CIPHER_SUITES_CONFIG =
"kafkastore.ssl.cipher.suites";
public static final String KAFKASTORE_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG =
"kafkastore.ssl.endpoint.identification.algorithm";
protected static final String KAFKASTORE_CONNECTION_URL_DOC =
"Zookeeper url for the Kafka cluster";
protected static final String SCHEMAREGISTRY_ZK_NAMESPACE_DOC =
Expand Down Expand Up @@ -111,10 +144,43 @@ public class SchemaRegistryConfig extends RestConfig {
+ "forward (latest registered schema can read data produced by the new schema), "
+ "full (new schema is backward and forward compatible with latest registered schema)";
protected static final String MASTER_ELIGIBILITY_DOC =
"If true, this node can participate in master election. In a multi-colo setup, turn this off"
+ "for clusters in the slave data center.";
"If true, this node can participate in master election. In a multi-colo setup, turn this off "
+ "for clusters in the slave data center.";
protected static final String KAFKASTORE_SECURITY_PROTOCOL_DOC =
"The security protocol to use when connecting with Kafka, the underlying persistent storage. "
+ "Values can be `PLAINTEXT` or `SSL`.";
protected static final String KAFKASTORE_SSL_TRUSTSTORE_LOCATION_DOC =
"The location of the SSL trust store file.";
protected static final String KAFKASTORE_SSL_TRUSTSTORE_PASSWORD_DOC =
"The password to access the trust store.";
protected static final String KAFAKSTORE_SSL_TRUSTSTORE_TYPE_DOC =
"The file format of the trust store.";
protected static final String KAFKASTORE_SSL_TRUSTMANAGER_ALGORITHM_DOC =
"The algorithm used by the trust manager factory for SSL connections.";
protected static final String KAFKASTORE_SSL_KEYSTORE_LOCATION_DOC =
"The location of the SSL keystore file.";
protected static final String KAFKASTORE_SSL_KEYSTORE_PASSWORD_DOC =
"The password to access the keystore.";
protected static final String KAFAKSTORE_SSL_KEYSTORE_TYPE_DOC =
"The file format of the keystore.";
protected static final String KAFKASTORE_SSL_KEYMANAGER_ALGORITHM_DOC =
"The algorithm used by key manager factory for SSL connections.";
protected static final String KAFKASTORE_SSL_KEY_PASSWORD_DOC =
"The password of the key contained in the keystore.";
protected static final String KAFAKSTORE_SSL_ENABLED_PROTOCOLS_DOC =
"Protocols enabled for SSL connections.";
protected static final String KAFAKSTORE_SSL_PROTOCOL_DOC =
"The SSL protocol used.";
protected static final String KAFAKSTORE_SSL_PROVIDER_DOC =
"The name of the security provider used for SSL.";
protected static final String KAFKASTORE_SSL_CIPHER_SUITES_DOC =
"A list of cipher suites used for SSL.";
protected static final String KAFKASTORE_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC =
"The endpoint identification algorithm to validate the server hostname using the server certificate.";
private static final String COMPATIBILITY_DEFAULT = "backward";
private static final String METRICS_JMX_PREFIX_DEFAULT_OVERRIDE = "kafka.schema.registry";

// TODO: move to Apache's ConfigDef
private static final ConfigDef config;
static {
config = baseConfigDef()
Expand Down Expand Up @@ -152,7 +218,52 @@ public class SchemaRegistryConfig extends RestConfig {
ConfigDef.Importance.MEDIUM, MASTER_ELIGIBILITY_DOC)
.defineOverride(METRICS_JMX_PREFIX_CONFIG, ConfigDef.Type.STRING,
METRICS_JMX_PREFIX_DEFAULT_OVERRIDE, ConfigDef.Importance.LOW,
METRICS_JMX_PREFIX_DOC);
METRICS_JMX_PREFIX_DOC)
.define(KAFKASTORE_SECURITY_PROTOCOL_CONFIG, ConfigDef.Type.STRING,
KAFKASTORE_SECURITY_PROTOCOL_PLAINTEXT, ConfigDef.Importance.MEDIUM,
KAFKASTORE_SECURITY_PROTOCOL_DOC)

@ijuma Ismael Juma (ijuma) May 6, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The importance of this is MEDIUM for the producer and consumer config, is it intentional that we set it to HIGH here? It probably makes sense, thinking about it, but just wanted to double-check.

@ijuma Ismael Juma (ijuma) May 6, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The importance for a few other configs has also been changed. It would be interesting to have an explanation of the reasoning (I don't doubt that it's sound :)).

@ijuma Ismael Juma (ijuma) May 6, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

By the way, if we agree on the changes in priority, then you can submit a PR to change it in Kafka too.

@alexlod Alex L (alexlod) May 9, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I made up my own priorities based on what felt good. I don't feel strongly about keeping them as is vs being consistent with Kafka. I'll make them consistent with Kafka unless I hear otherwise.

@ijuma Ismael Juma (ijuma) May 9, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm fine either way, I think thinking through priorities would probably be a good thing, but given our tight deadline, it probably makes sense to just do what Kafka does.

.define(KAFKASTORE_SSL_TRUSTSTORE_LOCATION_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.HIGH,
KAFKASTORE_SSL_TRUSTSTORE_LOCATION_DOC)
.define(KAFKASTORE_SSL_TRUSTSTORE_PASSWORD_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.HIGH,
KAFKASTORE_SSL_TRUSTSTORE_PASSWORD_DOC)
.define(KAFKASTORE_SSL_TRUSTSTORE_TYPE_CONFIG, ConfigDef.Type.STRING,
"JKS", ConfigDef.Importance.MEDIUM,
KAFAKSTORE_SSL_TRUSTSTORE_TYPE_DOC)
.define(KAFKASTORE_SSL_TRUSTMANAGER_ALGORITHM_CONFIG, ConfigDef.Type.STRING,
"PKIX", ConfigDef.Importance.LOW,
KAFKASTORE_SSL_TRUSTMANAGER_ALGORITHM_DOC)
.define(KAFKASTORE_SSL_KEYSTORE_LOCATION_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.HIGH,
KAFKASTORE_SSL_KEYSTORE_LOCATION_DOC)
.define(KAFKASTORE_SSL_KEYSTORE_PASSWORD_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.HIGH,
KAFKASTORE_SSL_KEYSTORE_PASSWORD_DOC)
.define(KAFKASTORE_SSL_KEYSTORE_TYPE_CONFIG, ConfigDef.Type.STRING,
"JKS", ConfigDef.Importance.MEDIUM,
KAFAKSTORE_SSL_KEYSTORE_TYPE_DOC)
.define(KAFKASTORE_SSL_KEYMANAGER_ALGORITHM_CONFIG, ConfigDef.Type.STRING,
"SunX509", ConfigDef.Importance.LOW,
KAFKASTORE_SSL_KEYMANAGER_ALGORITHM_DOC)
.define(KAFKASTORE_SSL_KEY_PASSWORD_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.HIGH,
KAFKASTORE_SSL_KEY_PASSWORD_DOC)
.define(KAFKASTORE_SSL_ENABLED_PROTOCOLS_CONFIG, ConfigDef.Type.STRING,
"TLSv1.2,TLSv1.1,TLSv1", ConfigDef.Importance.MEDIUM,
KAFAKSTORE_SSL_ENABLED_PROTOCOLS_DOC)
.define(KAFKASTORE_SSL_PROTOCOL_CONFIG, ConfigDef.Type.STRING,
"TLS", ConfigDef.Importance.MEDIUM,
KAFAKSTORE_SSL_PROTOCOL_DOC)
.define(KAFKASTORE_SSL_PROVIDER_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.MEDIUM,
KAFAKSTORE_SSL_PROVIDER_DOC)
.define(KAFKASTORE_SSL_CIPHER_SUITES_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.LOW,
KAFKASTORE_SSL_CIPHER_SUITES_DOC)
.define(KAFKASTORE_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, ConfigDef.Type.STRING,
"", ConfigDef.Importance.LOW,
KAFKASTORE_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC);
}
private final AvroCompatibilityLevel compatibilityType;

Expand Down
62 changes: 58 additions & 4 deletions core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,14 @@
import kafka.admin.RackAwareMode;
import kafka.cluster.EndPoint;
import kafka.server.ConfigType;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.security.JaasUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -77,6 +79,7 @@ public class KafkaStore<K, V> implements Store<K, V> {
// messages with this key
private final K noopKey;
private volatile long lastWrittenOffset = -1L;
private final SchemaRegistryConfig config;

public KafkaStore(SchemaRegistryConfig config,
StoreUpdateHandler<K, V> storeUpdateHandler,
Expand Down Expand Up @@ -107,6 +110,8 @@ public KafkaStore(SchemaRegistryConfig config,

this.bootstrapBrokers = KafkaStore.getBrokerEndpoints(
JavaConversions.seqAsJavaList(this.brokerSeq));

this.config = config;
}

@Override
Expand All @@ -128,14 +133,17 @@ public void init() throws StoreInitializationException {
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
org.apache.kafka.common.serialization.ByteArraySerializer.class);
props.put(ProducerConfig.RETRIES_CONFIG, 0); // Producer should not retry

addSslConfigsToClientProperties(this.config, props);

producer = new KafkaProducer<byte[],byte[]>(props);

// start the background thread that subscribes to the Kafka topic and applies updates.
// the thread must be created after the schema topic has been created.
this.kafkaTopicReader =
new KafkaStoreReaderThread<>(this.bootstrapBrokers, topic, groupId,
this.storeUpdateHandler, serializer, this.localStore,
this.noopKey);
this.noopKey, this.config);
this.kafkaTopicReader.start();

try {
Expand All @@ -151,6 +159,51 @@ public void init() throws StoreInitializationException {
}
}

public static void addSslConfigsToClientProperties(SchemaRegistryConfig config, Properties props) {
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SECURITY_PROTOCOL_CONFIG));
if (config.getString(SchemaRegistryConfig.KAFKASTORE_SECURITY_PROTOCOL_CONFIG).equals(
SchemaRegistryConfig.KAFKASTORE_SECURITY_PROTOCOL_SSL)) {
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_TRUSTSTORE_LOCATION_CONFIG));
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_TRUSTSTORE_PASSWORD_CONFIG));
props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_TRUSTSTORE_TYPE_CONFIG));
props.put(SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_TRUSTMANAGER_ALGORITHM_CONFIG));
putIfNotEmptyString(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_KEYSTORE_LOCATION_CONFIG), props);
putIfNotEmptyString(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_KEYSTORE_PASSWORD_CONFIG), props);
props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_KEYSTORE_TYPE_CONFIG));
props.put(SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_KEYMANAGER_ALGORITHM_CONFIG));
putIfNotEmptyString(SslConfigs.SSL_KEY_PASSWORD_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_KEY_PASSWORD_CONFIG), props);
putIfNotEmptyString(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_ENABLED_PROTOCOLS_CONFIG), props);
props.put(SslConfigs.SSL_PROTOCOL_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_PROTOCOL_CONFIG));
putIfNotEmptyString(SslConfigs.SSL_PROVIDER_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_PROVIDER_CONFIG), props);
putIfNotEmptyString(SslConfigs.SSL_CIPHER_SUITES_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_CIPHER_SUITES_CONFIG), props);
putIfNotEmptyString(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,
config.getString(SchemaRegistryConfig.KAFKASTORE_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG), props);
}
}

// helper method to only add a property if its not the empty string. This is required
// because some Kafka client configs expect a null default value, yet ConfigDef doesn't
// support null default values.
private static void putIfNotEmptyString(String parameter, String value, Properties props) {
if (!value.trim().isEmpty()) {
props.put(parameter, value);
}
}

private void createSchemaTopic() throws StoreInitializationException {
if (AdminUtils.topicExists(zkUtils, topic)) {
verifySchemaTopic();
Expand Down Expand Up @@ -185,19 +238,20 @@ static String getBrokerEndpoints(List<Broker> brokerList) {
for(EndPoint ep : JavaConversions.asJavaCollection(broker.endPoints().values())) {
String connectionString = ep.connectionString();

if (connectionString.startsWith("PLAINTEXT://")) {
if (connectionString.startsWith(SchemaRegistryConfig.KAFKASTORE_SECURITY_PROTOCOL_SSL + "://")
|| connectionString.startsWith(SchemaRegistryConfig.KAFKASTORE_SECURITY_PROTOCOL_PLAINTEXT + "://")) {
if (sb.length() > 0) {
sb.append(",");
}
sb.append(connectionString);
} else {
log.warn("Ignoring non-plaintext Kafka endpoint: " + connectionString);
log.warn("Ignoring non-plaintext and non-SSL Kafka endpoint: " + connectionString);
}
}
}

if (sb.length() == 0) {
throw new ConfigException("Only plaintext Kafka endpoints are supported and " +
throw new ConfigException("Only plaintext and SSL Kafka endpoints are supported and " +
"none are configured.");
}

Expand Down
9 changes: 8 additions & 1 deletion core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaStoreReaderThread.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,16 @@
*/
package io.confluent.kafka.schemaregistry.storage;

import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
Expand Down Expand Up @@ -73,7 +76,8 @@ public KafkaStoreReaderThread(String bootstrapBrokers,
StoreUpdateHandler<K, V> storeUpdateHandler,
Serializer<K, V> serializer,
Store<K, V> localStore,
K noopKey) {
K noopKey,
SchemaRegistryConfig config) {
super("kafka-store-reader-thread-" + topic, false); // this thread is not interruptible
offsetUpdateLock = new ReentrantLock();
offsetReachedThreshold = offsetUpdateLock.newCondition();
Expand All @@ -95,6 +99,9 @@ public KafkaStoreReaderThread(String bootstrapBrokers,
org.apache.kafka.common.serialization.ByteArrayDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
org.apache.kafka.common.serialization.ByteArrayDeserializer.class);

KafkaStore.addSslConfigsToClientProperties(config, consumerProps);

this.consumer = new KafkaConsumer<>(consumerProps);

List<PartitionInfo> partitions = this.consumer.partitionsFor(this.topic);
Expand Down
Loading