stefan-egli commented on code in PR #43:
URL:
https://github.com/apache/sling-org-apache-sling-event/pull/43#discussion_r2084869684
##########
src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java:
##########
@@ -122,8 +122,8 @@ public class JobManagerImpl
@Reference
private StatisticsManager statisticsManager;
-
-
+
+
Review Comment:
unnecessary change
##########
src/main/java/org/apache/sling/event/impl/jobs/queues/JobQueueImpl.java:
##########
@@ -169,7 +169,7 @@ public static JobQueueImpl createQueue(final String name,
* @param cache The job cache
* @param outdatedQueue
*/
- private JobQueueImpl(final String name,
+ public JobQueueImpl(final String name,
Review Comment:
same comment here as done on another comment (but outside this review) : I
would try to avoid making this class public and go via a test-accessor, if
possible/feasible...
##########
src/main/java/org/apache/sling/event/impl/jobs/queues/JobQueueImpl.java:
##########
@@ -468,6 +476,10 @@ public void close() {
* Periodic maintenance
*/
public void maintain() {
+ if (!services.configuration.isJobProcessingEnabled()) {
+ logger.debug("Job processing is disabled, skipping maintenance for
queue {}", queueName);
+ return;
+ }
Review Comment:
having third thought on the need for this as mentioned in another comment:
if `isActive` contains the readiness condition value, then this method
shouldn't actually be called at all in that case ...
##########
src/main/java/org/apache/sling/event/impl/jobs/queues/JobQueueImpl.java:
##########
@@ -789,5 +814,13 @@ QueueJobCache getCache() {
OutdatedJobQueueInfo getOutdatedJobQueueInfo() {
return new OutdatedJobQueueInfo(available, maxParallel, drainage);
}
+
+ public Map<String, JobHandler> getProcessingJobsLists() {
+ return processingJobsLists;
+ }
+
+ public boolean isRunning() {
+ return running;
+ }
Review Comment:
(if possible I'd try to go via test accessor and make these package
protected)
##########
src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java:
##########
@@ -199,6 +202,50 @@ static JobManagerConfiguration
newForTest(ResourceResolverFactory resourceResolv
/** The topology capabilities. */
private volatile TopologyCapabilities topologyCapabilities;
+ /** The condition that determines if job processing is enabled. */
+ @Reference(
+ target =
"(osgi.condition.id=org.apache.sling.event.jobs.processing.enabled)",
+ cardinality = ReferenceCardinality.OPTIONAL,
+ policy = ReferencePolicy.DYNAMIC,
+ policyOption = ReferencePolicyOption.GREEDY
+ )
+ private volatile Condition jobProcessingEnabledCondition;
+
+ /**
+ * Handle binding of the job processing condition.
+ * @param condition The condition being bound
+ */
+ protected void bindJobProcessingEnabledCondition(final Condition
condition) {
+ this.jobProcessingEnabledCondition = condition;
+ // If condition becomes true, trigger maintenance to start processing
jobs
+ if (condition != null) {
+ notifyListeners();
+ }
+ }
+
+ /**
+ * Handle unbinding of the job processing condition.
+ * @param condition The condition being unbound
+ */
+ protected void unbindJobProcessingEnabledCondition(final Condition
condition) {
+ if (this.jobProcessingEnabledCondition == condition) {
+ this.jobProcessingEnabledCondition = null;
Review Comment:
maybe add a `log.info` about readiness condition having been removed?
##########
src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java:
##########
@@ -623,7 +670,7 @@ void doHandleTopologyEvent(final TopologyEvent event) {
public void addListener(final ConfigurationChangeListener service) {
synchronized ( this.listeners ) {
this.listeners.add(service);
- service.configurationChanged(this.topologyCapabilities != null);
+ service.configurationChanged(this.topologyCapabilities != null &&
isJobProcessingEnabled());
Review Comment:
(same here as above comment)
##########
src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java:
##########
@@ -565,7 +612,7 @@ private void notifyListeners() {
synchronized ( this.listeners ) {
final TopologyCapabilities caps = this.topologyCapabilities;
for(final ConfigurationChangeListener l : this.listeners) {
- l.configurationChanged(caps != null);
+ l.configurationChanged(caps != null &&
isJobProcessingEnabled());
Review Comment:
I'm wondering if this is really correct.
There seem to be two different use cases - JobSchedulerImpl and
QueueManager. The former actually should be based on the former definition of
`active` while the latter probably needs to include this flag indeed.
One way could be that it wouldn't be so much `isJobProcessingEnabled` but
rather `isJobStartingEnabled` - in which case the `configurationChanged` call
here wouldn't include that flag - but the QueueManager.configurationChanged
would. In which case actually - the QueueManager would set `isActive` based on
this flag as well - thus removing the need for protecting the
`JobQueueImpl.maintain` - as that is already protected by that `isActive` .....
##########
src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java:
##########
@@ -95,7 +95,7 @@
})
public class JobManagerImpl
implements JobManager, EventHandler, Runnable {
-
+
Review Comment:
unnecessary change
##########
src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java:
##########
@@ -391,11 +391,24 @@ public void configurationChanged(final boolean active) {
if ( active ) {
fullTopicScan();
} else {
+ // Stop all running jobs before restarting
+ stopAllJobs();
Review Comment:
interesting one.
I think this could be potentially dangerous - as it would stop the jobs in
case a new instance joins the topology - which might or might not be what we
want.
I wouldn't do this here in general - but only if the readiness condition
became false
--
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: dev-unsubscr...@sling.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
