cameronlee314 commented on a change in pull request #1039: SAMZA-2197: [Scala cleanup] Convert TaskConfig to Java URL: https://github.com/apache/samza/pull/1039#discussion_r290847071
########## File path: samza-core/src/main/java/org/apache/samza/config/TaskConfig.java ########## @@ -0,0 +1,298 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.samza.config; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.apache.commons.lang3.StringUtils; +import org.apache.samza.Partition; +import org.apache.samza.checkpoint.CheckpointManager; +import org.apache.samza.checkpoint.CheckpointManagerFactory; +import org.apache.samza.container.grouper.task.GroupByContainerCountFactory; +import org.apache.samza.metrics.MetricsRegistry; +import org.apache.samza.system.SystemStream; +import org.apache.samza.system.SystemStreamPartition; +import org.apache.samza.util.StreamUtil; +import org.apache.samza.util.Util; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Helper for accessing config values related to tasks. + */ +public class TaskConfig extends MapConfig { + public static final Logger LOGGER = LoggerFactory.getLogger(TaskConfig.class); + + // comma-separated list of system-streams + public static final String INPUT_STREAMS = "task.inputs"; + // window period in milliseconds + public static final String WINDOW_MS = "task.window.ms"; + static final long DEFAULT_WINDOW_MS = -1L; + // commit period in milliseconds + public static final String COMMIT_MS = "task.commit.ms"; + static final long DEFAULT_COMMIT_MS = 60000L; + // how long to wait for a clean shutdown + public static final String TASK_SHUTDOWN_MS = "task.shutdown.ms"; + static final long DEFAULT_TASK_SHUTDOWN_MS = 30000L; + // legacy config for specifying task class; replaced by SamzaApplication and app.class + public static final String TASK_CLASS = "task.class"; + // command builder to use for launching a Samza job + public static final String COMMAND_BUILDER = "task.command.class"; + // message chooser for controlling stream consumption + public static final String MESSAGE_CHOOSER_CLASS_NAME = "task.chooser.class"; + // define whether drop the messages or not when deserialization fails + public static final String DROP_DESERIALIZATION_ERRORS = "task.drop.deserialization.errors"; + // define whether drop the messages or not when serialization fails + public static final String DROP_SERIALIZATION_ERRORS = "task.drop.serialization.errors"; + // whether to ignore producer errors and drop the messages that failed to send + public static final String DROP_PRODUCER_ERRORS = "task.drop.producer.errors"; + // exceptions to ignore in process and window + public static final String IGNORED_EXCEPTIONS = "task.ignored.exceptions"; + // class name for task grouper + public static final String GROUPER_FACTORY = "task.name.grouper.factory"; + // max number of concurrent process for a AsyncStreamTask + public static final String MAX_CONCURRENCY = "task.max.concurrency"; + static final int DEFAULT_MAX_CONCURRENCY = 1; + // timeout period for triggering a callback + public static final String CALLBACK_TIMEOUT_MS = "task.callback.timeout.ms"; + static final long DEFAULT_CALLBACK_TIMEOUT_MS = -1L; + // to enable async commit in a AsyncStreamTask Review comment: If `task.max.concurrency` works with `StreamTask`, then it looks like this would too. I'll update it. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
