Minor fix to some config variable names and accessor methods.

Author: Prateek Maheshwari <[email protected]>

Reviewers: Jagadish<[email protected]>

Closes apache#840 from prateekm/fix-config-names

Author: prateekm

samza-core/src/main/scala/org/apache/samza/config/TaskConfig.scala

@@ -36,9 +36,9 @@ object TaskConfig {
   val LIFECYCLE_LISTENER = "task.lifecycle.listener.%s.class" // task.lifecycle.listener.li-generator.class
   val CHECKPOINT_MANAGER_FACTORY = TaskConfigJava.CHECKPOINT_MANAGER_FACTORY // class name to use when sending offset checkpoints
   val MESSAGE_CHOOSER_CLASS_NAME = "task.chooser.class"
-  val DROP_DESERIALIZATION_ERROR = "task.drop.deserialization.errors" // define whether drop the messages or not when deserialization fails
-  val DROP_SERIALIZATION_ERROR = "task.drop.serialization.errors" // define whether drop the messages or not when serialization fails
-  val DROP_PRODUCER_ERROR = "task.drop.producer.errors" // whether to ignore producer errors and drop the messages that failed to send
+  val DROP_DESERIALIZATION_ERRORS = "task.drop.deserialization.errors" // define whether drop the messages or not when deserialization fails
+  val DROP_SERIALIZATION_ERRORS = "task.drop.serialization.errors" // define whether drop the messages or not when serialization fails
+  val DROP_PRODUCER_ERRORS = "task.drop.producer.errors" // whether to ignore producer errors and drop the messages that failed to send
   val IGNORED_EXCEPTIONS = "task.ignored.exceptions" // exceptions to ignore in process and window
   val GROUPER_FACTORY = "task.name.grouper.factory" // class name for task grouper
   val MAX_CONCURRENCY = "task.max.concurrency" // max number of concurrent process for a AsyncStreamTask
@@ -115,11 +115,11 @@ class TaskConfig(config: Config) extends ScalaMapConfig(config) with Logging {
 
   def getMessageChooserClass = getOption(TaskConfig.MESSAGE_CHOOSER_CLASS_NAME)
 
-  def getDropDeserialization = getOption(TaskConfig.DROP_DESERIALIZATION_ERROR)
+  def getDropDeserializationErrors = getBoolean(TaskConfig.DROP_DESERIALIZATION_ERRORS, false)
 
-  def getDropSerialization = getOption(TaskConfig.DROP_SERIALIZATION_ERROR)
+  def getDropSerializationErrors = getBoolean(TaskConfig.DROP_SERIALIZATION_ERRORS, false)
 
-  def getDropProducerError = getBoolean(TaskConfig.DROP_PRODUCER_ERROR, false)
+  def getDropProducerErrors = getBoolean(TaskConfig.DROP_PRODUCER_ERRORS, false)
 
   def getPollIntervalMs = getOption(TaskConfig.POLL_INTERVAL_MS)
 

samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala

@@ -425,15 +425,8 @@ object SamzaContainer extends Logging {
     val offsetManager = OffsetManager(inputStreamMetadata, config, checkpointManager, systemAdmins, checkpointListeners, offsetManagerMetrics)
     info("Got offset manager: %s" format offsetManager)
 
-    val dropDeserializationError = config.getDropDeserialization match {
-      case Some(dropError) => dropError.toBoolean
-      case _ => false
-    }
-
-    val dropSerializationError = config.getDropSerialization match {
-      case Some(dropError) => dropError.toBoolean
-      case _ => false
-    }
+    val dropDeserializationError = config.getDropDeserializationErrors
+    val dropSerializationError = config.getDropSerializationErrors
 
     val pollIntervalMs = config
       .getPollIntervalMs

samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala

@@ -83,7 +83,7 @@ class KafkaSystemFactory extends SystemFactory with Logging {
       new ExponentialSleepStrategy(initialDelayMs = producerConfig.reconnectIntervalMs),
       getProducer,
       metrics,
-      dropProducerExceptions = config.getDropProducerError)
+      dropProducerExceptions = config.getDropProducerErrors)
   }
 
   def getAdmin(systemName: String, config: Config): SystemAdmin = {

samza-kafka/src/main/scala/org/apache/samza/system/kafka_deprecated/KafkaSystemFactory.scala

@@ -101,7 +101,7 @@ class KafkaSystemFactory extends SystemFactory with Logging {
       new ExponentialSleepStrategy(initialDelayMs = producerConfig.reconnectIntervalMs),
       getProducer,
       metrics,
-      dropProducerExceptions = config.getDropProducerError)
+      dropProducerExceptions = config.getDropProducerErrors)
   }
 
   def getAdmin(systemName: String, config: Config): SystemAdmin = {

samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java

@@ -279,7 +279,7 @@ protected Config getConfig() {
       throw new SamzaException("can not read the config", e);
     }
     // Make system producer drop producer errors for StreamAppender
-    config = new MapConfig(config, ImmutableMap.of(TaskConfig.DROP_PRODUCER_ERROR(), "true"));
+    config = new MapConfig(config, ImmutableMap.of(TaskConfig.DROP_PRODUCER_ERRORS(), "true"));
 
     return config;
   }

samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java

@@ -300,7 +300,7 @@ protected Config getConfig() {
       throw new SamzaException("can not read the config", e);
     }
     // Make system producer drop producer errors for StreamAppender
-    config = new MapConfig(config, ImmutableMap.of(TaskConfig.DROP_PRODUCER_ERROR(), "true"));
+    config = new MapConfig(config, ImmutableMap.of(TaskConfig.DROP_PRODUCER_ERRORS(), "true"));
 
     return config;
   }

samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java

@@ -206,7 +206,7 @@ private Map<String, String> buildStreamApplicationConfigMap(String systemName, S
         .put(JobConfig.JOB_NAME(), appName)
         .put(JobConfig.JOB_ID(), appId)
         .put(TaskConfigJava.TASK_SHUTDOWN_MS, TASK_SHUTDOWN_MS)
-        .put(TaskConfig.DROP_PRODUCER_ERROR(), "true")
+        .put(TaskConfig.DROP_PRODUCER_ERRORS(), "true")
         .put(JobConfig.JOB_DEBOUNCE_TIME_MS(), JOB_DEBOUNCE_TIME_MS)
         .put(JobConfig.MONITOR_PARTITION_CHANGE_FREQUENCY_MS(), "1000")
         .build();