You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by GitBox <gi...@apache.org> on 2021/08/18 21:50:01 UTC
[GitHub] [samza] cameronlee314 opened a new pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
cameronlee314 opened a new pull request #1520:
URL: https://github.com/apache/samza/pull/1520
Issues: Construction of stream monitors (`StreamPartitionCountMonitor`, `StreamRegexMonitor`) is directly implemented in the job coordinators. Therefore, it is hard to reuse the code across multiple places.
Changes:
1. Extract construction logic into factory classes for `StreamPartitionCountMonitor` and `StreamRegexMonitor`.
2. Update job coordinator classes to use the factories.
Tests:
`./gradlew build`
API changes and upgrade instructions: N/A
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
mynameborat commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698666392
##########
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:
The only reason being is when read `streamPartitionCountMonitorFactory.buildInputStramPartitionCountMonitor(..)` sounds redundant.
I don't mind verbosity but felt it was redundant. I'd prefer conciseness if we can achieve the same intent.
Fair enough, it doesn't have to know about it here. But the regex factory apparently does care about the fact whether its rewritten config vs original config.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 merged pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 merged pull request #1520:
URL: https://github.com/apache/samza/pull/1520
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
mynameborat commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698666660
##########
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:
I didn't mean specifically about the StreamPartitionCountMonitor.
Rather, you have two parameters needed for the factory passed through constructor and two parameters through `build(..)` method.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698709630
##########
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:
I looked more closely at what `ClusterBasedJobCoordinator` was using before. It was using the config from before calculating the job model. It seems like it should ideally be using the config from after calculating job model, but I would prefer to not change current logic in this PR, so that this PR can purely stay as a refactoring PR. I will leave the variable name as `config`, since other variable names like `finalConfig` could be inconsistent with the current logic. I will also create a ticket around this more general problem of having too many "config stages"; most objects should only receive the final config.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 merged pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 merged pull request #1520:
URL: https://github.com/apache/samza/pull/1520
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698671144
##########
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:
Oh, I see. The dependencies passed to the factory constructor are always the same across all calls to the factory, while the dependencies passed to `build` are different across each call to the factory.
By passing some dependencies through the constructor to the factory, the caller of `build` doesn't have to worry about them.
##########
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:
Ok, I do prefer the conciseness as well, so I will change it back. If we do need differentiation in the future, we can just change the method name.
I suppose it won't hurt to use the name `finalConfig` to help be a little more explicit, but I still don't think the stage of the config should be part of the contract between classes (unless they the classes are directly dealing with config rewriting), because that introduces an ordering dependency. Most classes should be written in a way that is agnostic to where they are being used.
##########
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:
I looked more closely at what `ClusterBasedJobCoordinator` was using before. It was using the config from before calculating the job model. It seems like it should ideally be using the config from after calculating job model, but I would prefer to not change current logic in this PR, so that this PR can purely stay as a refactoring PR. I will leave the variable name as `config`, since other variable names like `finalConfig` could be inconsistent with the current logic. I will also create a ticket around this more general problem of having too many "config stages"; most objects should only receive the final config.
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
+ }
+
+ public StreamPartitionCountMonitor build(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.");
Review comment:
I prefer to keep as-is, since `StreamPartitionCountMonitorFactory` is always expected to return a monitor (e.g. there is no config to disable it). `StreamRegexMonitorFactory` is not always expected to return a monitor, because the config might disable 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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] alnzng commented on pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
alnzng commented on pull request #1520:
URL: https://github.com/apache/samza/pull/1520#issuecomment-906093341
Thanks for the changes. LGTM.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698738324
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
+ }
+
+ public StreamPartitionCountMonitor build(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.");
Review comment:
I prefer to keep as-is, since `StreamPartitionCountMonitorFactory` is always expected to return a monitor (e.g. there is no config to disable it). `StreamRegexMonitorFactory` is not always expected to return a monitor, because the config might disable 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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
mynameborat commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698817121
##########
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:
I didn't realize you are reusing factory instance to create more objects subsequently.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698680590
##########
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:
Ok, I do prefer the conciseness as well, so I will change it back. If we do need differentiation in the future, we can just change the method name.
I suppose it won't hurt to use the name `finalConfig` to help be a little more explicit, but I still don't think the stage of the config should be part of the contract between classes (unless they the classes are directly dealing with config rewriting), because that introduces an ordering dependency. Most classes should be written in a way that is agnostic to where they are being used.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] alnzng commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
alnzng commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r696288023
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
+ }
+
+ public StreamPartitionCountMonitor build(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.");
Review comment:
Just for API consistency between `StreamPartitionCountMonitorFactory` and `StreamRegexMonitorFactory`.
But I'm good if we keep current logic.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
mynameborat commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698666392
##########
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:
The only reason being is when read `streamPartitionCountMonitorFactory.buildInputStramPartitionCountMonitor(..)` sounds redundant.
I don't mind verbosity but felt it was redundant. I'd prefer conciseness if we can achieve the same intent.
Fair enough, it doesn't have to know about it here. But the regex factory apparently does care about the fact whether its rewritten config vs original config.
##########
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:
I didn't mean specifically about the StreamPartitionCountMonitor.
Rather, you have two parameters needed for the factory passed through constructor and two parameters through `build(..)` method.
##########
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:
I didn't realize you are reusing factory instance to create more objects subsequently.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r695943599
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
+ }
+
+ public StreamPartitionCountMonitor build(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.");
Review comment:
What is the benefit you see for changing it?
If this is changed, then someone has to guarantee that something else makes this check. Also, we would need to change the classes which use this method to handle Optional when they were not using Optional before.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] alnzng commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
alnzng commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r695383102
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
+ }
+
+ public StreamPartitionCountMonitor build(Config config, StreamPartitionCountMonitor.Callback callback) {
Review comment:
maybe `buildInputStreamPartitionCountMonitor()` is a better name?
Just from this factory name, it is generic enough which means it allows us to build StreamPartitionCountMonitor for output streams in the future.
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamRegexMonitorFactory.java
##########
@@ -0,0 +1,83 @@
+/*
+ * 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;
+
+
+public class StreamRegexMonitorFactory {
+ private static final Logger LOG = LoggerFactory.getLogger(StreamRegexMonitorFactory.class);
+
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamRegexMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
+ }
+
+ public Optional<StreamRegexMonitor> build(JobModel jobModel, StreamRegexMonitor.Callback callback) {
Review comment:
maybe `buildInputStreamRegexMonitor()` is a better name?
Just from this factory name, it is generic enough which means it allows us to build StreamRegexMonitor for output streams in the future.
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamRegexMonitorFactory.java
##########
@@ -0,0 +1,83 @@
+/*
+ * 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;
+
+
+public class StreamRegexMonitorFactory {
Review comment:
Can you add some Javadoc for the public API?
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamPartitionCountMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
+ }
+
+ public StreamPartitionCountMonitor build(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.");
Review comment:
I know these are the existing codes. But I wonder if we can just `return Optional.empty()` instead of throwing SamzaException. IIUC, The job planning process happen before and it already checks the input streams.
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
Review comment:
Can you add some Javadoc for public API?
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r696040015
##########
File path: samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+
+public class StreamPartitionCountMonitorFactory {
Review comment:
To make sure we are on the same page, this isn't a "public" API in the sense that Samza jobs will be using it (it is in `samza-core`). It is "public" in the sense that it is a public class in Java.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] cameronlee314 commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r698671144
##########
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:
Oh, I see. The dependencies passed to the factory constructor are always the same across all calls to the factory, while the dependencies passed to `build` are different across each call to the factory.
By passing some dependencies through the constructor to the factory, the caller of `build` doesn't have to worry about them.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a change in pull request #1520: SAMZA-2672: Extract creation of stream monitors into separate classes for reuse
Posted by GitBox <gi...@apache.org>.
mynameborat commented on a change in pull request #1520:
URL: https://github.com/apache/samza/pull/1520#discussion_r696296834
##########
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:
Would prefer calling this just `build` since the factory is not minting different types of `StreamPartitionCountMonitor` as in not like `SystemFactory` which mints admin, consumer and producer.
##########
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:
Why have two different ways of passing the dependencies needed for building the partition count monitor?
##########
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);
+
+ private final StreamMetadataCache streamMetadataCache;
+ private final MetricsRegistry metrics;
+
+ public StreamRegexMonitorFactory(StreamMetadataCache streamMetadataCache, MetricsRegistry metrics) {
+ this.streamMetadataCache = streamMetadataCache;
+ this.metrics = metrics;
Review comment:
Do we have any invariants to check on these dependencies (like non-null etc)?
##########
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:
Same comment as above class.
##########
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:
This isn't related to your change but it will be useful to rename the parameter `config` to more explicit name such as `finalConfig`, `originalConfig` etc to just make it more readable.
--
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: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org