ProduceService and ConsumerService should not allow certain properties in the properties override config (linkedin#34)

Author: WarrenGreen

services/src/main/java/com/linkedin/kmf/services/ConsumeService.java

@@ -16,9 +16,11 @@
 import com.linkedin.kmf.consumer.BaseConsumerRecord;
 import com.linkedin.kmf.consumer.NewConsumer;
 import com.linkedin.kmf.consumer.OldConsumer;
+import com.linkedin.kmf.services.configs.ProduceServiceConfig;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.Measurable;
 import org.apache.kafka.common.metrics.MetricConfig;
@@ -47,6 +49,8 @@
 public class ConsumeService implements Service {
   private static final Logger LOG = LoggerFactory.getLogger(ConsumeService.class);
   private static final String METRIC_GROUP_NAME = "consume-service";
+  private static final String[] NONOVERRIDABLE_PROPERTIES = new String[]{ ConsumeServiceConfig.BOOTSTRAP_SERVERS_CONFIG,
+                                                                          ConsumeServiceConfig.ZOOKEEPER_CONNECT_CONFIG };
 
   private final String _name;
   private final ConsumeMetrics _sensors;
@@ -59,7 +63,8 @@ public class ConsumeService implements Service {
 
   public ConsumeService(Map<String, Object> props, String name) throws Exception {
     _name = name;
-    Map consumerPropsOverride = (Map) props.get(ConsumeServiceConfig.CONSUMER_PROPS_CONFIG);
+    Map consumerPropsOverride = props.containsKey(ConsumeServiceConfig.CONSUMER_PROPS_CONFIG) ?
+        (Map) props.get(ConsumeServiceConfig.CONSUMER_PROPS_CONFIG) : new HashMap<>();
     ConsumeServiceConfig config = new ConsumeServiceConfig(props);
     String topic = config.getString(ConsumeServiceConfig.TOPIC_CONFIG);
     String zkConnect = config.getString(ConsumeServiceConfig.ZOOKEEPER_CONNECT_CONFIG);
@@ -70,6 +75,12 @@ public ConsumeService(Map<String, Object> props, String name) throws Exception {
     _latencyPercentileGranularityMs = config.getInt(ConsumeServiceConfig.LATENCY_PERCENTILE_GRANULARITY_MS_CONFIG);
     _running = new AtomicBoolean(false);
 
+    for (String property: NONOVERRIDABLE_PROPERTIES) {
+      if (consumerPropsOverride.containsKey(property)) {
+        throw new ConfigException("Override must not contain " + property + " config.");
+      }
+    }
+
     Properties consumerProps = new Properties();
 
     // Assign default config. This has the lowest priority.
@@ -94,8 +105,7 @@ public ConsumeService(Map<String, Object> props, String name) throws Exception {
     consumerProps.put("zookeeper.connect", zkConnect);
 
     // Assign config specified for consumer. This has the highest priority.
-    if (consumerPropsOverride != null)
-      consumerProps.putAll(consumerPropsOverride);
+    consumerProps.putAll(consumerPropsOverride);
 
     _consumer = (KMBaseConsumer) Class.forName(consumerClassName).getConstructor(String.class, Properties.class).newInstance(topic, consumerProps);
 

services/src/main/java/com/linkedin/kmf/services/ProduceService.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.Measurable;
 import org.apache.kafka.common.metrics.MetricConfig;
@@ -48,6 +49,8 @@
 public class ProduceService implements Service {
   private static final Logger LOG = LoggerFactory.getLogger(ProduceService.class);
   private static final String METRIC_GROUP_NAME = "produce-service";
+  private static final String[] NONOVERRIDABLE_PROPERTIES = new String[]{ ProduceServiceConfig.BOOTSTRAP_SERVERS_CONFIG,
+                                                                        ProduceServiceConfig.ZOOKEEPER_CONNECT_CONFIG };
 
   private final String _name;
   private final ProduceMetrics _sensors;
@@ -72,7 +75,6 @@ public class ProduceService implements Service {
 
   public ProduceService(Map<String, Object> props, String name) throws Exception {
     _name = name;
-    _producerPropsOverride = (Map) props.get(ProduceServiceConfig.PRODUCER_PROPS_CONFIG);
     ProduceServiceConfig config = new ProduceServiceConfig(props);
     _zkConnect = config.getString(ProduceServiceConfig.ZOOKEEPER_CONNECT_CONFIG);
     _brokerList = config.getString(ProduceServiceConfig.BOOTSTRAP_SERVERS_CONFIG);
@@ -86,6 +88,14 @@ public ProduceService(Map<String, Object> props, String name) throws Exception {
     _partitionNum = new AtomicInteger(0);
     _running = new AtomicBoolean(false);
     _nextIndexPerPartition = new ConcurrentHashMap<>();
+    _producerPropsOverride = props.containsKey(ProduceServiceConfig.PRODUCER_PROPS_CONFIG) ?
+      (Map) props.get(ProduceServiceConfig.PRODUCER_PROPS_CONFIG) : new HashMap<>();
+
+    for (String property: NONOVERRIDABLE_PROPERTIES) {
+      if (_producerPropsOverride.containsKey(property)) {
+        throw new ConfigException("Override must not contain " + property + " config.");
+      }
+    }
 
     double partitionsToBrokersRatio = config.getDouble(CommonServiceConfig.PARTITIONS_TO_BROKER_RATO_CONFIG);
     int topicReplicationFactor = config.getInt(ProduceServiceConfig.TOPIC_REPLICATION_FACTOR_CONFIG);
@@ -147,8 +157,7 @@ private void initializeProducer() throws Exception {
     producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _brokerList);
 
     // Assign config specified for producer. This has the highest priority.
-    if (_producerPropsOverride != null)
-      producerProps.putAll(_producerPropsOverride);
+    producerProps.putAll(_producerPropsOverride);
 
     _producer = (KMBaseProducer) Class.forName(_producerClassName).getConstructor(Properties.class).newInstance(producerProps);
     LOG.info("Producer is initialized.");