cameronlee314 commented on a change in pull request #1520: URL: https://github.com/apache/samza/pull/1520#discussion_r696875688
########## File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java ########## @@ -0,0 +1,55 @@ +/* + * 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.coordinator; + +import java.util.Set; +import org.apache.samza.SamzaException; +import org.apache.samza.config.Config; +import org.apache.samza.config.JobConfig; +import org.apache.samza.config.TaskConfig; +import org.apache.samza.metrics.MetricsRegistry; +import org.apache.samza.system.StreamMetadataCache; +import org.apache.samza.system.SystemStream; + + +/** + * Common place for reading configs and wiring {@link StreamPartitionCountMonitor}. + */ +public class StreamPartitionCountMonitorFactory { + private final StreamMetadataCache streamMetadataCache; + private final MetricsRegistry metrics; + + public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) { + this.streamMetadataCache = streamMetadataCache; + this.metrics = metrics; + } + + /** + * Build {@link StreamPartitionCountMonitor} for input streams. + */ + public StreamPartitionCountMonitor buildInputStreamPartitionCountMonitor(Config config, + StreamPartitionCountMonitor.Callback callback) { + Set<SystemStream> inputStreamsToMonitor = new TaskConfig(config).getAllInputStreams(); + if (inputStreamsToMonitor.isEmpty()) { + throw new SamzaException("Input streams to a job can not be empty."); + } + return new StreamPartitionCountMonitor(inputStreamsToMonitor, this.streamMetadataCache, this.metrics, + new JobConfig(config).getMonitorPartitionChangeFrequency(), callback); Review comment: The direct dependencies of `StreamPartitionCountMonitor` are more complex, so the factory is a place to do things like read configs. I suppose this factory looks a bit less necessary since it doesn't do that much, but if it needs to get extended in the future to have more checks, then this becomes a good common place. For example, `StreamRegexMonitorFactory` has more things to check, so this pattern provides a good common place for those kinds of things. ########## File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java ########## @@ -0,0 +1,55 @@ +/* + * 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.coordinator; + +import java.util.Set; +import org.apache.samza.SamzaException; +import org.apache.samza.config.Config; +import org.apache.samza.config.JobConfig; +import org.apache.samza.config.TaskConfig; +import org.apache.samza.metrics.MetricsRegistry; +import org.apache.samza.system.StreamMetadataCache; +import org.apache.samza.system.SystemStream; + + +/** + * Common place for reading configs and wiring {@link StreamPartitionCountMonitor}. + */ +public class StreamPartitionCountMonitorFactory { + private final StreamMetadataCache streamMetadataCache; + private final MetricsRegistry metrics; + + public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) { + this.streamMetadataCache = streamMetadataCache; + this.metrics = metrics; + } + + /** + * Build {@link StreamPartitionCountMonitor} for input streams. + */ + public StreamPartitionCountMonitor buildInputStreamPartitionCountMonitor(Config config, Review comment: 1. I originally had it as `build` but @alnzng suggested changing it. I didn't have a strong preference, but I don't mind the verbosity. Do you feel strongly about changing it back? 2. There is just a single `Config` in this method, so do we need to add semantics to the variable name about "original" vs. "final"? The method contract is to build the monitor using some config, but should this method care about whether it was "original" vs. "final" in the scope of the rest of the job? ########## File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamRegexMonitorFactory.java ########## @@ -0,0 +1,90 @@ +/* + * 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.coordinator; + +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.regex.Pattern; +import org.apache.samza.SamzaException; +import org.apache.samza.config.JobConfig; +import org.apache.samza.job.model.JobModel; +import org.apache.samza.job.model.JobModelUtil; +import org.apache.samza.metrics.MetricsRegistry; +import org.apache.samza.system.StreamMetadataCache; +import org.apache.samza.system.SystemStream; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Common place for reading configs and wiring {@link StreamRegexMonitor}. + */ +public class StreamRegexMonitorFactory { + private static final Logger LOG = LoggerFactory.getLogger(StreamRegexMonitorFactory.class); Review comment: There are a non-trivial amount of checks (e.g. configs) to build the `StreamRegexMonitor`. We shouldn't copy that code to multiple places. Also, I didn't want to put this code into a `StreamRegexMonitor` constructor since that doesn't follow dependency injection (which makes it harder to test). -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
