MINOR: adding schema registry configs
*/
package io.confluent.examples.streams.kafka;
-import io.confluent.examples.streams.zookeeper.ZooKeeperEmbedded;
-import io.confluent.kafka.schemaregistry.RestApp;
-import io.confluent.kafka.schemaregistry.avro.AvroCompatibilityLevel;
-import kafka.Kafka;
-import kafka.server.KafkaConfig$;
-import kafka.utils.ZkUtils;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.security.JaasUtils;
import org.apache.kafka.test.TestCondition;
@@ -35,6 +29,13 @@ import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
+import io.confluent.examples.streams.zookeeper.ZooKeeperEmbedded;
+import io.confluent.kafka.schemaregistry.RestApp;
+import io.confluent.kafka.schemaregistry.avro.AvroCompatibilityLevel;
+import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
+import kafka.server.KafkaConfig$;
+import kafka.utils.ZkUtils;
+
/**
* Runs an in-memory, "embedded" Kafka cluster with 1 ZooKeeper instance, 1 Kafka broker, and 1
* Confluent Schema Registry instance.
@@ -46,6 +47,10 @@ public class EmbeddedSingleNodeKafkaCluster extends ExternalResource {
private static final String KAFKA_SCHEMAS_TOPIC = "_schemas";
private static final String AVRO_COMPATIBILITY_TYPE = AvroCompatibilityLevel.NONE.name;
+ private static final String KAFKASTORE_OPERATION_TIMEOUT_MS = "10000";
+ private static final String KAFKASTORE_DEBUG = "true";
+ private static final String KAFKASTORE_INIT_TIMEOUT = "90000";
+
private ZooKeeperEmbedded zookeeper;
private ZkUtils zkUtils = null;
private KafkaEmbedded broker;
@@ -92,7 +97,14 @@ public class EmbeddedSingleNodeKafkaCluster extends ExternalResource {
log.debug("Kafka instance is running at {}, connected to ZooKeeper at {}",
broker.brokerList(), broker.zookeeperConnect());
+ Properties schemaRegistryProps = new Properties();
+
+ schemaRegistryProps.put(SchemaRegistryConfig.KAFKASTORE_TIMEOUT_CONFIG, KAFKASTORE_OPERATION_TIMEOUT_MS);
+ schemaRegistryProps.put(SchemaRegistryConfig.DEBUG_CONFIG, KAFKASTORE_DEBUG);
+ schemaRegistryProps.put(SchemaRegistryConfig.KAFKASTORE_INIT_TIMEOUT_CONFIG, KAFKASTORE_INIT_TIMEOUT);
+
schemaRegistry = new RestApp(0, zookeeperConnect(), KAFKA_SCHEMAS_TOPIC, AVRO_COMPATIBILITY_TYPE);
+ schemaRegistry.addConfigs(schemaRegistryProps);
schemaRegistry.start();
running = true;
}
