SLING-11831 - Allow setting job properties for custom job state

bnd.bnd

@@ -19,7 +19,7 @@ Export-Package: \
     org.apache.sling.event.api
 
 -includeresource:\
-    org.apache.sling.event.api-1.0.0.jar=org.apache.sling.event.api-1.0.0.jar;lib:=true,\
+    org.apache.sling.event.api-1.0.2.jar=org.apache.sling.event.api-1.0.2.jar;lib:=true,\
     @quartz-[0-9.]*.jar!/org/quartz/CronExpression.class,\
     @quartz-[0-9.]*.jar!/org/quartz/ValueSet.class,\
     @org.apache.sling.commons.osgi-[0-9.]*.jar!/org/apache/sling/commons/osgi/PropertiesUtil.class,\

pom.xml

@@ -119,6 +119,11 @@
             <groupId>org.osgi</groupId>
             <artifactId>org.osgi.service.component.annotations</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.jetbrains</groupId>
+            <artifactId>annotations</artifactId>
+            <scope>compile</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.sling</groupId>
             <artifactId>org.apache.sling.discovery.api</artifactId>
@@ -269,7 +274,7 @@
         <dependency>
             <groupId>org.apache.sling</groupId>
             <artifactId>org.apache.sling.event.api</artifactId>
-            <version>1.0.0</version>
+            <version>1.0.2</version>
         </dependency>
 
         <dependency>

src/main/java/org/apache/sling/event/impl/jobs/queues/JobExecutionContextImpl.java

@@ -24,6 +24,7 @@
 import org.apache.sling.event.jobs.Job;
 import org.apache.sling.event.jobs.consumer.JobExecutionContext;
 import org.apache.sling.event.jobs.consumer.JobExecutionResult;
+import org.jetbrains.annotations.NotNull;
 
 /**
  * Implementation of the job execution context passed to
@@ -85,7 +86,23 @@ public void updateProgress(final long eta) {
     }
 
     @Override
-    public void log(final String message, Object... args) {
+    public void setProperty(@NotNull final String name, final Object value) {
+        if ( name == null ) {
+            throw new IllegalArgumentException("Name must not be null");
+        }
+        if ( value == null ) {
+            throw new IllegalArgumentException("Value must not be null");
+        }
+        if ( name.startsWith("slingevent:") || name.startsWith(":slingevent:")) {
+            throw new IllegalArgumentException("Property name must not start with slingevent: or :slingevent: " + name);
+        }
+
+        handler.getJob().setProperty(name, value);
+        handler.persistJobProperties(name);
+    }
+
+    @Override
+    public void log(@NotNull final String message, Object... args) {
         final int logMaxCount = handler.getProgressLogMaxCount();
         handler.persistJobProperties(handler.getJob().log(logMaxCount, message, args));
     }

src/main/java/org/apache/sling/event/impl/jobs/tasks/CleanUpTask.java

@@ -150,6 +150,10 @@ private void historyCleanUpRemovedJobs(Calendar since) {
             HistoryCleanUpTask.cleanup(
                     since,
                     resolver,
+                    /**
+                     * We use a dummy context here as we are running it as a
+                     * scheduled task and not as a job.
+                     */
                     new JobExecutionContext() {
                         @Override
                         public void asyncProcessingFinished(JobExecutionResult result) {
@@ -181,6 +185,11 @@ public void log(String message, Object... args) {
 
                         }
 
+                        @Override
+                        public void setProperty(String name, Object value) {
+
+                        }
+
                         @Override
                         public ResultBuilder result() {
                             return new ResultBuilderImpl();
@@ -436,7 +445,7 @@ private void cleanUpInstanceIdFolders(final TopologyCapabilities caps, final Str
                 for(final Resource r : toDelete) {
                     if ( caps.isActive() ) {
                         resolver.delete(r);
-                        resolver.commit();    
+                        resolver.commit();
                     }
                 }
             }

src/test/java/org/apache/sling/event/impl/jobs/queues/JobExecutionContextImplTest.java

@@ -0,0 +1,163 @@
+/*
+ * 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.sling.event.impl.jobs.queues;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.math.BigInteger;
+import java.util.HashMap;
+
+import org.apache.sling.api.resource.Resource;
+import org.apache.sling.api.resource.ResourceResolver;
+import org.apache.sling.api.resource.ValueMap;
+import org.apache.sling.commons.scheduler.Scheduler;
+import org.apache.sling.commons.threads.ThreadPoolManager;
+import org.apache.sling.event.impl.jobs.JobConsumerManager;
+import org.apache.sling.event.impl.jobs.JobHandler;
+import org.apache.sling.event.impl.jobs.JobImpl;
+import org.apache.sling.event.impl.jobs.JobManagerImpl;
+import org.apache.sling.event.impl.jobs.config.InternalQueueConfiguration;
+import org.apache.sling.event.impl.jobs.config.JobManagerConfiguration;
+import org.apache.sling.event.impl.jobs.config.QueueConfigurationManager;
+import org.apache.sling.event.impl.jobs.config.QueueConfigurationManager.QueueInfo;
+import org.apache.sling.event.impl.jobs.config.TopologyCapabilities;
+import org.apache.sling.event.impl.jobs.stats.StatisticsManager;
+import org.apache.sling.event.jobs.Job;
+import org.apache.sling.event.jobs.JobManager;
+import org.apache.sling.event.jobs.consumer.JobExecutionContext;
+import org.apache.sling.event.jobs.consumer.JobExecutionResult;
+import org.apache.sling.event.jobs.consumer.JobExecutor;
+import org.apache.sling.event.jobs.jmx.QueuesMBean;
+import org.apache.sling.testing.mock.sling.junit.SlingContext;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.slf4j.LoggerFactory;
+
+import com.codahale.metrics.MetricRegistry;
+
+public class JobExecutionContextImplTest {
+
+    @Rule
+    public SlingContext context = new SlingContext();
+
+    private JobManager jobManager;
+    private JobManagerConfiguration configuration;
+
+    @Before
+    public void setUp() {
+        configuration = createMockJobManagerConfiguration();
+
+        QueueConfigurationManager queueConfigMgr = mock(QueueConfigurationManager.class);
+        QueueInfo info = new QueueInfo();
+        info.queueConfiguration = new InternalQueueConfiguration();
+        when(queueConfigMgr.getQueueInfo(anyString())).thenReturn(info);
+        when(configuration.getQueueConfigurationManager()).thenReturn(queueConfigMgr);
+
+        TopologyCapabilities capabilities = mock(TopologyCapabilities.class);
+        JobConsumerManager jobConsumerManager = mock(JobConsumerManager.class);
+        QueueManager qManager = mock(QueueManager.class);
+        ThreadPoolManager threadPoolManager = mock(ThreadPoolManager.class);
+        MetricRegistry metric = mock(MetricRegistry.class);
+        StatisticsManager statisticsManager = mock(StatisticsManager.class);
+        QueuesMBean queuesMBean = mock(QueuesMBean.class);
+        Scheduler scheduler = mock(Scheduler.class);
+
+        context.registerService(JobManagerConfiguration.class, configuration);
+        context.registerService(TopologyCapabilities.class, capabilities);
+        context.registerService(QueueConfigurationManager.class, queueConfigMgr);
+        context.registerService(MetricRegistry.class, metric);
+        context.registerService(QueueManager.class, qManager);
+        context.registerService(JobConsumerManager.class, jobConsumerManager);
+        context.registerService(ThreadPoolManager.class, threadPoolManager);
+        context.registerService(StatisticsManager.class, statisticsManager);
+        context.registerService(QueuesMBean.class, queuesMBean);
+        context.registerService(Scheduler.class, scheduler);
+        context.registerService(JobExecutor.class, new TestJobExecutor(), new HashMap<String, Object>() {{
+            put(JobExecutor.PROPERTY_TOPICS, "test");
+        }});
+
+        jobManager = new JobManagerImpl();
+        context.registerInjectActivateService(jobManager, new HashMap<String, Object>());
+    }
+
+    @Test
+    public void testSetProperty() {
+        // Create a job - it will be written to the mock jcr
+        Job job = jobManager.addJob("test", null);
+
+        // Process the job
+        JobExecutor je = new TestJobExecutor();
+        je.process(job, new JobExecutionContextImpl(new JobHandler((JobImpl) job, je, configuration), null));
+
+        // Retrieve the custom property
+        assertEquals("testValue", job.getProperty("test", String.class));
+
+        final String testValue;
+        Iterable<Resource> resources = context.resourceResolver().getResource("/var/eventing/jobs/assigned").getChildren();
+        ValueMap props = resources.iterator().next().adaptTo(ValueMap.class);
+        testValue = props.get("test", String.class);
+        assertEquals("testValue", testValue);
+    }
+
+    public class TestJobExecutor implements JobExecutor {
+
+        @Override
+        public JobExecutionResult process(Job job, JobExecutionContext context) {
+            context.setProperty("test", "testValue");
+            return context.result().message("TEST").succeeded();
+        }
+    }
+
+    private JobManagerConfiguration createMockJobManagerConfiguration() {
+        JobManagerConfiguration jobManagerConfig = mock(JobManagerConfiguration.class);
+
+        String jobsPath = "/var/eventing";
+
+        when(jobManagerConfig.getUniqueId(anyString())).then(new Answer<String>() {
+            @Override
+            public String answer(InvocationOnMock invocation) throws Throwable {
+                byte [] digest = java.security.MessageDigest.getInstance("md5").digest(String.valueOf(Math.random()).getBytes("UTF-8"));
+                BigInteger bigInt = new BigInteger(1, digest);
+                String hashtext = bigInt.toString(16);
+                return hashtext + "_" + String.valueOf((int)(Math.random()* 1000000));
+            }
+        });
+        when(jobManagerConfig.getScheduledJobsPath(false)).thenReturn(jobsPath + "/scheduled-jobs");
+        when(jobManagerConfig.getUniquePath(eq(null), anyString(), anyString(), eq(null))).then(
+            new Answer<String>() {
+                @Override
+                public String answer(InvocationOnMock invocation) throws Throwable {
+                    String jobNodePath = jobsPath + "/jobs/assigned/" + invocation.getArgument(2);
+                    return jobNodePath;
+                }
+            }
+        );
+        when(jobManagerConfig.getAuditLogger()).thenReturn(LoggerFactory.getLogger("org.apache.sling.event.jobs.audit"));
+        ResourceResolver resolver = context.resourceResolver();
+        when(jobManagerConfig.createResourceResolver()).thenReturn(resolver);
+        return jobManagerConfig;
+    }
+}