From ce7b7850f765dc45ee1bbbaa2b05897e0b8ec1de Mon Sep 17 00:00:00 2001
From: Tiago Dolphine <tiagodolphine@gmail.com>
Date: Thu, 31 Oct 2019 16:49:49 -0300
Subject: [PATCH] Initial Job Service implementation (#1)

* Initial Job Service implementation

Initial Job Service implementation

inserting swagger

* Apply PR comments and some code refactoring, test

* Fix JobScheduler when cancelling job

* Fixing cancel already scheduled job when re-scheduling
---
 .../java/org/kie/kogito/jobs/api/Job.java     |  43 +++++-
 .../org/kie/kogito/jobs/api/JobBuilder.java   |  79 ++++++++++
 addons/jobs/jobs-service/.gitignore           |   2 +
 addons/jobs/jobs-service/pom.xml              | 137 +++++++++++++++++-
 .../src/main/docker/Dockerfile.jvm            |  31 ++++
 .../src/main/docker/Dockerfile.native         |  22 +++
 .../service/converters/HttpConverters.java    |  41 ++++++
 .../service/executor/HttpJobExecutor.java     |  92 ++++++++++++
 .../jobs/service/executor/JobExecutor.java    |  26 ++++
 .../service/executor/RetryJobExecutor.java    |  29 ++++
 .../service/json/JacksonConfiguration.java    |  53 +++++++
 .../jobs/service/model/ErrorResponse.java     |  30 ++++
 .../service/model/HTTPRequestCallback.java    | 135 +++++++++++++++++
 .../jobs/service/model/ScheduledJob.java      |  55 +++++++
 .../repository/ReactiveJobRepository.java     |  36 +++++
 .../impl/InMemoryJobRepository.java           |  94 ++++++++++++
 .../resource/CallbackResourceTest.java        |  51 +++++++
 .../jobs/service/resource/JobResource.java    |  89 ++++++++++++
 .../resource/error/DefaultErrorMapper.java    |  44 ++++++
 .../scheduler/BaseTimerJobScheduler.java      |  98 +++++++++++++
 .../jobs/service/scheduler/JobScheduler.java  |  26 ++++
 .../scheduler/ReactiveJobScheduler.java       |  30 ++++
 .../scheduler/impl/VertxJobScheduler.java     |  66 +++++++++
 .../src/main/resources/application.properties |  44 ++++++
 .../service/resource/JobResourceTest.java     | 111 ++++++++++++++
 25 files changed, 1459 insertions(+), 5 deletions(-)
 create mode 100644 addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/JobBuilder.java
 create mode 100644 addons/jobs/jobs-service/src/main/docker/Dockerfile.jvm
 create mode 100644 addons/jobs/jobs-service/src/main/docker/Dockerfile.native
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/converters/HttpConverters.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/HttpJobExecutor.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/JobExecutor.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/RetryJobExecutor.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/json/JacksonConfiguration.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ErrorResponse.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/HTTPRequestCallback.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ScheduledJob.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/ReactiveJobRepository.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/impl/InMemoryJobRepository.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/CallbackResourceTest.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/JobResource.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/error/DefaultErrorMapper.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/BaseTimerJobScheduler.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/JobScheduler.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/ReactiveJobScheduler.java
 create mode 100644 addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/impl/VertxJobScheduler.java
 create mode 100644 addons/jobs/jobs-service/src/main/resources/application.properties
 create mode 100644 addons/jobs/jobs-service/src/test/java/org/kie/kogito/jobs/service/resource/JobResourceTest.java

diff --git a/addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/Job.java b/addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/Job.java
index b9031157aac..f59312e959c 100644
--- a/addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/Job.java
+++ b/addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/Job.java
@@ -16,16 +16,17 @@
 package org.kie.kogito.jobs.api;
 
 import java.time.ZonedDateTime;
+import java.util.Objects;
 
 /**
- * Job describes the actual entity that should be scheduled and executed 
+ * Job describes the actual entity that should be scheduled and executed
  * upon given expiration time. The job requires following information
  * <ul>
  *  <li>id - unique UUID based identifier</li>
  *  <li>expirationTime - the time when this job should be executed</li>
  *  <li>callbackEndpoint - the callback endpoint (http/https) that will be invoked upon expiration</li>
  * </ul>
- * 
+ *
  * On top of that there are additional meta data that points the job to the owner - such as process instance.
  * <ul>
  *  <li>processInstanceId - process instance that owns the job</li>
@@ -49,6 +50,20 @@ public class Job {
     private String processId;
     private String rootProcessId;
 
+    public Job() {
+    }
+
+    public Job(String id, ZonedDateTime expirationTime, Integer priority, String callbackEndpoint, String processInstanceId, String rootProcessInstanceId, String processId, String rootProcessId) {
+        this.id = id;
+        this.expirationTime = expirationTime;
+        this.priority = priority;
+        this.callbackEndpoint = callbackEndpoint;
+        this.processInstanceId = processInstanceId;
+        this.rootProcessInstanceId = rootProcessInstanceId;
+        this.processId = processId;
+        this.rootProcessId = rootProcessId;
+    }
+
     public String getId() {
         return id;
     }
@@ -113,6 +128,30 @@ public void setRootProcessId(String rootProcessId) {
         this.rootProcessId = rootProcessId;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof Job)) {
+            return false;
+        }
+        Job job = (Job) o;
+        return Objects.equals(getId(), job.getId()) &&
+                Objects.equals(getExpirationTime(), job.getExpirationTime()) &&
+                Objects.equals(getPriority(), job.getPriority()) &&
+                Objects.equals(getCallbackEndpoint(), job.getCallbackEndpoint()) &&
+                Objects.equals(getProcessInstanceId(), job.getProcessInstanceId()) &&
+                Objects.equals(getRootProcessInstanceId(), job.getRootProcessInstanceId()) &&
+                Objects.equals(getProcessId(), job.getProcessId()) &&
+                Objects.equals(getRootProcessId(), job.getRootProcessId());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getId(), getExpirationTime(), getPriority(), getCallbackEndpoint(), getProcessInstanceId(), getRootProcessInstanceId(), getProcessId(), getRootProcessId());
+    }
+
     @Override
     public String toString() {
         return "Job [id=" + id + ", expirationTime=" + expirationTime + ", priority=" + priority + ", callbackEndpoint=" + callbackEndpoint + ", processInstanceId=" + processInstanceId + ", rootProcessInstanceId=" +
diff --git a/addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/JobBuilder.java b/addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/JobBuilder.java
new file mode 100644
index 00000000000..50501070bc0
--- /dev/null
+++ b/addons/jobs/jobs-api/src/main/java/org/kie/kogito/jobs/api/JobBuilder.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.api;
+
+import java.time.ZonedDateTime;
+
+public class JobBuilder {
+
+    private String id;
+    private ZonedDateTime expirationTime;
+    private Integer priority;
+    private String callbackEndpoint;
+    private String processInstanceId;
+    private String rootProcessInstanceId;
+    private String processId;
+    private String rootProcessId;
+
+    public JobBuilder id(String id) {
+        this.id = id;
+        return this;
+    }
+
+    public JobBuilder expirationTime(ZonedDateTime expirationTime) {
+        this.expirationTime = expirationTime;
+        return this;
+    }
+
+    public JobBuilder priority(Integer priority) {
+        this.priority = priority;
+        return this;
+    }
+
+    public JobBuilder callbackEndpoint(String callbackEndpoint) {
+        this.callbackEndpoint = callbackEndpoint;
+        return this;
+    }
+
+    public JobBuilder processInstanceId(String processInstanceId) {
+        this.processInstanceId = processInstanceId;
+        return this;
+    }
+
+    public JobBuilder rootProcessInstanceId(String rootProcessInstanceId) {
+        this.rootProcessInstanceId = rootProcessInstanceId;
+        return this;
+    }
+
+    public JobBuilder processId(String processId) {
+        this.processId = processId;
+        return this;
+    }
+
+    public JobBuilder rootProcessId(String rootProcessId) {
+        this.rootProcessId = rootProcessId;
+        return this;
+    }
+
+    public Job build() {
+        return new Job(id, expirationTime, priority, callbackEndpoint, processInstanceId, rootProcessInstanceId, processId, rootProcessId);
+    }
+
+    public static JobBuilder builder(){
+        return new JobBuilder();
+    }
+}
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/.gitignore b/addons/jobs/jobs-service/.gitignore
index 09322dd7f2a..bc4ac54309a 100644
--- a/addons/jobs/jobs-service/.gitignore
+++ b/addons/jobs/jobs-service/.gitignore
@@ -8,4 +8,6 @@ target/
 /*.ipr
 /*.iws
 *.iml
+*.log
+
 
diff --git a/addons/jobs/jobs-service/pom.xml b/addons/jobs/jobs-service/pom.xml
index d3315c2051a..2384c53636a 100644
--- a/addons/jobs/jobs-service/pom.xml
+++ b/addons/jobs/jobs-service/pom.xml
@@ -1,4 +1,6 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<?xml version="1.0"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
   <modelVersion>4.0.0</modelVersion>
   <parent>
     <groupId>org.kie.kogito</groupId>
@@ -7,5 +9,134 @@
   </parent>
   <artifactId>jobs-service</artifactId>
   <name>Jobs Service</name>
-  <description>Jobs Service (timers and async jobs)</description>
-</project>
\ No newline at end of file
+  <description>Jobs Service (Timers and Async Jobs)</description>
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>io.quarkus</groupId>
+        <artifactId>quarkus-bom</artifactId>
+        <version>${version.io.quarkus}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+  <dependencies>
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>jobs-api</artifactId>
+      <version>8.0.0-SNAPSHOT</version>
+    </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-vertx-web</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-vertx</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-web-client</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.smallrye.reactive</groupId>
+      <artifactId>smallrye-axle-web-client</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-resteasy-jackson</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.datatype</groupId>
+      <artifactId>jackson-datatype-jsr310</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-smallrye-openapi</artifactId>
+    </dependency>
+
+    <!-- Testing -->
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-junit5</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.rest-assured</groupId>
+      <artifactId>rest-assured</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+  </dependencies>
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>io.quarkus</groupId>
+        <artifactId>quarkus-maven-plugin</artifactId>
+        <executions>
+          <execution>
+            <goals>
+              <goal>build</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <artifactId>maven-compiler-plugin</artifactId>
+      </plugin>
+      <plugin>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <systemProperties>
+            <java.util.logging.manager>org.jboss.logmanager.LogManager</java.util.logging.manager>
+          </systemProperties>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+  <profiles>
+    <profile>
+      <id>native</id>
+      <activation>
+        <property>
+          <name>native</name>
+        </property>
+      </activation>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-maven-plugin</artifactId>
+            <executions>
+              <execution>
+                <goals>
+                  <goal>native-image</goal>
+                </goals>
+                <configuration>
+                  <enableHttpUrlHandler>true</enableHttpUrlHandler>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+          <plugin>
+            <artifactId>maven-failsafe-plugin</artifactId>
+            <executions>
+              <execution>
+                <goals>
+                  <goal>integration-test</goal>
+                  <goal>verify</goal>
+                </goals>
+                <configuration>
+                  <systemProperties>
+                    <native.image.path>${project.build.directory}/${project.build.finalName}-runner</native.image.path>
+                  </systemProperties>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+</project>
diff --git a/addons/jobs/jobs-service/src/main/docker/Dockerfile.jvm b/addons/jobs/jobs-service/src/main/docker/Dockerfile.jvm
new file mode 100644
index 00000000000..0939b39ef2b
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/docker/Dockerfile.jvm
@@ -0,0 +1,31 @@
+####
+# This Dockerfile is used in order to build a container that runs the Quarkus application in JVM mode
+#
+# Before building the docker image run:
+#
+# mvn package
+#
+# Then, build the image with:
+#
+# docker build -f src/main/docker/Dockerfile.jvm -t quarkus/jobs-service-jvm .
+#
+# Then run the container using:
+#
+# docker run -i --rm -p 8080:8080 quarkus/jobs-service-jvm
+#
+###
+FROM fabric8/java-alpine-openjdk8-jre
+ENV JAVA_OPTIONS="-Dquarkus.http.host=0.0.0.0 -Djava.util.logging.manager=org.jboss.logmanager.LogManager"
+ENV AB_ENABLED=jmx_exporter
+COPY target/lib/* /deployments/lib/
+COPY target/*-runner.jar /deployments/app.jar
+EXPOSE 8080
+
+# run with user 1001 and be prepared for be running in OpenShift too
+RUN adduser -G root --no-create-home --disabled-password 1001 \
+  && chown -R 1001 /deployments \
+  && chmod -R "g+rwX" /deployments \
+  && chown -R 1001:root /deployments
+USER 1001
+
+ENTRYPOINT [ "/deployments/run-java.sh" ]
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/main/docker/Dockerfile.native b/addons/jobs/jobs-service/src/main/docker/Dockerfile.native
new file mode 100644
index 00000000000..368d28e801e
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/docker/Dockerfile.native
@@ -0,0 +1,22 @@
+####
+# This Dockerfile is used in order to build a container that runs the Quarkus application in native (no JVM) mode
+#
+# Before building the docker image run:
+#
+# mvn package -Pnative -Dnative-image.docker-build=true
+#
+# Then, build the image with:
+#
+# docker build -f src/main/docker/Dockerfile.native -t quarkus/jobs-service .
+#
+# Then run the container using:
+#
+# docker run -i --rm -p 8080:8080 quarkus/jobs-service
+#
+###
+FROM registry.access.redhat.com/ubi8/ubi-minimal
+WORKDIR /work/
+COPY target/*-runner /work/application
+RUN chmod 775 /work
+EXPOSE 8080
+CMD ["./application", "-Dquarkus.http.host=0.0.0.0"]
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/converters/HttpConverters.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/converters/HttpConverters.java
new file mode 100644
index 00000000000..dcd128e3049
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/converters/HttpConverters.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.converters;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+
+import javax.enterprise.context.ApplicationScoped;
+
+import io.vertx.core.http.HttpMethod;
+import org.kie.kogito.jobs.service.model.HTTPRequestCallback;
+
+@ApplicationScoped
+public class HttpConverters {
+
+    public HttpMethod convertHttpMethod(HTTPRequestCallback.HTTPMethod method) {
+        return HttpMethod.valueOf(method.name());
+    }
+
+    public URL convertURL(String url){
+        try {
+            return new URL(url);
+        } catch (MalformedURLException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/HttpJobExecutor.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/HttpJobExecutor.java
new file mode 100644
index 00000000000..d32212f7d61
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/HttpJobExecutor.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.executor;
+
+import java.net.URL;
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+
+import javax.annotation.PostConstruct;
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+import io.vertx.axle.core.Vertx;
+import io.vertx.axle.ext.web.client.WebClient;
+import org.kie.kogito.jobs.api.Job;
+import org.kie.kogito.jobs.service.converters.HttpConverters;
+import org.kie.kogito.jobs.service.model.HTTPRequestCallback;
+import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@ApplicationScoped
+public class HttpJobExecutor implements JobExecutor {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(HttpJobExecutor.class);
+
+    @Inject
+    private Vertx vertx;
+
+    private WebClient client;
+
+    @Inject
+    private HttpConverters httpConverters;
+
+    @Inject
+    private ReactiveJobRepository jobRepository;
+
+    @PostConstruct
+    void initialize() {
+        this.client = WebClient.create(vertx);
+    }
+
+    private CompletionStage<Boolean> executeCallback(HTTPRequestCallback request) {
+        LOGGER.info("Executing callback {}", request);
+        final URL url = httpConverters.convertURL(request.getUrl());
+        return client.request(httpConverters.convertHttpMethod(request.getMethod()),
+                              url.getPort(),
+                              url.getHost(),
+                              url.getPath())
+                .send()
+                .thenApplyAsync(response -> Optional
+                        .ofNullable(response.statusCode())
+                        .filter(new Integer(200)::equals)
+                        .map(code -> Boolean.TRUE)
+                        .orElse(Boolean.FALSE));
+    }
+
+    @Override
+    public CompletionStage<Job> execute(Job job) {
+        //Using just POST method for now
+        final HTTPRequestCallback callback = HTTPRequestCallback.builder()
+                .url(job.getCallbackEndpoint())
+                .method(HTTPRequestCallback.HTTPMethod.POST)
+                .build();
+
+        return executeCallback(callback)
+                .thenApply(result -> {
+                    LOGGER.info("Response of executed job {} {}", result, job);
+                    jobRepository.delete(job.getId());
+                    return job;
+                })
+                //handle error
+                .exceptionally(ex -> {
+                    LOGGER.error("Error executing job " + job, ex);
+                    return job;
+                });
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/JobExecutor.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/JobExecutor.java
new file mode 100644
index 00000000000..f5f6f440cfd
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/JobExecutor.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.executor;
+
+import java.util.concurrent.CompletionStage;
+
+import org.kie.kogito.jobs.api.Job;
+
+public interface JobExecutor {
+
+    CompletionStage<Job> execute(Job job);
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/RetryJobExecutor.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/RetryJobExecutor.java
new file mode 100644
index 00000000000..b74a118f965
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/executor/RetryJobExecutor.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.executor;
+
+import java.util.concurrent.CompletionStage;
+
+import org.kie.kogito.jobs.api.Job;
+
+public class RetryJobExecutor implements JobExecutor {
+
+    @Override
+    public CompletionStage<Job> execute(Job job) {
+        return null;
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/json/JacksonConfiguration.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/json/JacksonConfiguration.java
new file mode 100644
index 00000000000..f4246295732
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/json/JacksonConfiguration.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.json;
+
+import java.util.Optional;
+
+import javax.annotation.PostConstruct;
+import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.inject.Produces;
+import javax.enterprise.inject.Specializes;
+import javax.inject.Singleton;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import io.quarkus.resteasy.jackson.runtime.ObjectMapperProducer;
+
+@ApplicationScoped
+@Specializes
+public class JacksonConfiguration extends ObjectMapperProducer {
+
+    private ObjectMapper objectMapper;
+
+    @PostConstruct
+    private ObjectMapper init() {
+        objectMapper = new ObjectMapper();
+        objectMapper.registerModule(new JavaTimeModule());
+        objectMapper.findAndRegisterModules();
+        objectMapper.disable(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE);
+        return objectMapper;
+    }
+
+    @Singleton
+    @Produces
+    @Override
+    public ObjectMapper objectMapper() {
+        return Optional.ofNullable(objectMapper).orElse(init());
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ErrorResponse.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ErrorResponse.java
new file mode 100644
index 00000000000..56a7ff40385
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ErrorResponse.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.model;
+
+public class ErrorResponse {
+
+    private String message;
+
+    public ErrorResponse(String message) {
+        this.message = message;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+}
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/HTTPRequestCallback.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/HTTPRequestCallback.java
new file mode 100644
index 00000000000..9676c57c62c
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/HTTPRequestCallback.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.model;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.StringJoiner;
+
+public class HTTPRequestCallback {
+
+    public enum HTTPMethod {
+        GET,
+        POST,
+        PUT,
+        DELETE,
+        PATCH,
+        OPTIONS,
+        HEAD,
+        TRACE,
+        CONNECT
+    }
+
+    private String url;
+    private HTTPMethod method;
+    private Map<String, String> headers;
+    private String body;
+
+    public HTTPRequestCallback(String url, HTTPMethod method, Map<String, String> headers, String body) {
+        this.url = url;
+        this.method = method;
+        this.headers = headers;
+        this.body = body;
+    }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public HTTPMethod getMethod() {
+        return method;
+    }
+
+    public Map<String, String> getHeaders() {
+        return headers;
+    }
+
+    public String getBody() {
+        return body;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof HTTPRequestCallback)) {
+            return false;
+        }
+        HTTPRequestCallback that = (HTTPRequestCallback) o;
+        return Objects.equals(getUrl(), that.getUrl()) &&
+                getMethod() == that.getMethod() &&
+                Objects.equals(getHeaders(), that.getHeaders()) &&
+                Objects.equals(getBody(), that.getBody());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getUrl(), getMethod(), getHeaders(), getBody());
+    }
+
+    @Override
+    public String toString() {
+        return new StringJoiner(", ", HTTPRequestCallback.class.getSimpleName() + "[", "]")
+                .add("url='" + url + "'")
+                .add("method=" + method)
+                .add("headers=" + headers)
+                .add("body='" + body + "'")
+                .toString();
+    }
+
+    public static Builder builder(){
+        return new Builder();
+    }
+
+    public static class Builder {
+
+        private String url;
+        private HTTPMethod method;
+        private Map<String, String> headers;
+        private String body;
+
+        public Builder url(String url) {
+            this.url = url;
+            return this;
+        }
+
+        public Builder method(String method) {
+            this.method = HTTPMethod.valueOf(method);
+            return this;
+        }
+
+        public Builder method(HTTPMethod method) {
+            this.method = method;
+            return this;
+        }
+
+        public Builder headers(Map<String, String> headers) {
+            this.headers = headers;
+            return this;
+        }
+
+        public Builder body(String body) {
+            this.body = body;
+            return this;
+        }
+
+        public HTTPRequestCallback build() {
+            return new HTTPRequestCallback(url, method, headers, body);
+        }
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ScheduledJob.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ScheduledJob.java
new file mode 100644
index 00000000000..9cd33debbbb
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/model/ScheduledJob.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.model;
+
+import java.util.Optional;
+import java.util.StringJoiner;
+
+import org.kie.kogito.jobs.api.Job;
+
+public class ScheduledJob {
+
+    private Job job;
+    private String scheduledId;
+    private Integer retries;
+
+    public ScheduledJob(Job job, String scheduledId) {
+        this.job = job;
+        this.scheduledId = scheduledId;
+    }
+
+    public Job getJob() {
+        return job;
+    }
+
+    public String getScheduledId() {
+        return scheduledId;
+    }
+
+    public Integer getRetries() {
+        return retries;
+    }
+
+    @Override
+    public String toString() {
+        return new StringJoiner(", ", ScheduledJob.class.getSimpleName() + "[", "]")
+                .add("job=" + Optional.ofNullable(job).map(Job::getId).orElse(null))
+                .add("scheduledId='" + scheduledId + "'")
+                .add("retries=" + retries)
+                .toString();
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/ReactiveJobRepository.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/ReactiveJobRepository.java
new file mode 100644
index 00000000000..39d243af21b
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/ReactiveJobRepository.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.repository;
+
+import java.time.ZonedDateTime;
+import java.util.concurrent.CompletionStage;
+
+import org.kie.kogito.jobs.service.model.ScheduledJob;
+import org.reactivestreams.Publisher;
+
+public interface ReactiveJobRepository {
+
+    CompletionStage<ScheduledJob> save(ScheduledJob job);
+
+    CompletionStage<ScheduledJob> get(String id);
+
+    Publisher<ScheduledJob> getByTime(ZonedDateTime expirationTime);
+
+    CompletionStage<Boolean> exists(String id);
+
+    CompletionStage<ScheduledJob> delete(String id);
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/impl/InMemoryJobRepository.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/impl/InMemoryJobRepository.java
new file mode 100644
index 00000000000..27a82c8edba
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/repository/impl/InMemoryJobRepository.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.repository.impl;
+
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+import io.vertx.core.Vertx;
+import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
+import org.kie.kogito.jobs.service.model.ScheduledJob;
+import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
+import org.reactivestreams.Publisher;
+
+@ApplicationScoped
+public class InMemoryJobRepository implements ReactiveJobRepository {
+
+    private final Map<String, ScheduledJob> jobMap = new ConcurrentHashMap<>();
+    private final Map<ZonedDateTime, List<String>> jobTimeMap = new ConcurrentHashMap<>();
+
+    @Inject
+    private Vertx vertx;
+
+    @Override
+    public CompletionStage<ScheduledJob> save(ScheduledJob job) {
+        return runAsync(() -> {
+            jobMap.put(job.getJob().getId(), job);
+            jobTimeMap.putIfAbsent(getExpirationTime(job), new ArrayList<>());
+            jobTimeMap.get(getExpirationTime(job)).add(job.getJob().getId());
+            return job;
+        });
+    }
+
+    @Override
+    public CompletionStage<ScheduledJob> get(String key) {
+        return runAsync(() -> jobMap.get(key));
+    }
+
+    @Override
+    public CompletionStage<Boolean> exists(String key) {
+        return runAsync(() -> jobMap.containsKey(key));
+    }
+
+    @Override
+    public CompletionStage<ScheduledJob> delete(String key) {
+        return runAsync(() -> {
+            Optional.ofNullable(jobTimeMap.get(getExpirationTime(jobMap.get(key))))
+                    .map(jobs-> jobs.remove(key));
+            return jobMap.remove(key);
+        });
+    }
+
+    private ZonedDateTime getExpirationTime(ScheduledJob job) {
+        return job.getJob().getExpirationTime();
+    }
+
+    @Override
+    public Publisher<ScheduledJob> getByTime(ZonedDateTime expirationTime) {
+        return ReactiveStreams.fromIterable(jobTimeMap.getOrDefault(expirationTime, Collections.emptyList()))
+                .map(this::get)
+                .flatMapCompletionStage(j -> j)
+                .buildRs();
+    }
+
+    private <T> CompletionStage<T> runAsync(Supplier<T> function) {
+        final CompletableFuture<T> future = new CompletableFuture<>();
+        vertx.runOnContext((v) -> future.complete(function.get()));
+        return future;
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/CallbackResourceTest.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/CallbackResourceTest.java
new file mode 100644
index 00000000000..865388bda50
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/CallbackResourceTest.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.resource;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Path("/callback")
+public class CallbackResourceTest {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(CallbackResourceTest.class);
+
+    @POST
+    @Produces(MediaType.TEXT_PLAIN)
+    @Consumes(MediaType.APPLICATION_JSON)
+    public CompletionStage<String> post() {
+        LOGGER.debug("post received");
+        return CompletableFuture.completedFuture("Post Success");
+    }
+
+    @GET
+    @Produces(MediaType.TEXT_PLAIN)
+    public CompletionStage<String> get() {
+        LOGGER.debug("get received");
+        return CompletableFuture.completedFuture("Get Success");
+    }
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/JobResource.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/JobResource.java
new file mode 100644
index 00000000000..dca5301916c
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/JobResource.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.resource;
+
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+
+import javax.inject.Inject;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
+import org.kie.kogito.jobs.api.Job;
+import org.kie.kogito.jobs.service.model.ScheduledJob;
+import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
+import org.kie.kogito.jobs.service.scheduler.impl.VertxJobScheduler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Path("/job")
+public class JobResource {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(JobResource.class);
+
+    @Inject
+    private VertxJobScheduler scheduler;
+
+    @Inject
+    private ReactiveJobRepository reactiveJobRepository;
+
+    @POST
+    @Produces(MediaType.APPLICATION_JSON)
+    @Consumes(MediaType.APPLICATION_JSON)
+    public CompletionStage<Job> create(Job job) {
+        LOGGER.debug("REST create {}", job);
+        final CompletionStage<Job> response = ReactiveStreams.fromPublisher(scheduler.schedule(job))
+                .map(ScheduledJob::getJob)
+                .findFirst()
+                .run()
+                .thenApply(j -> j.orElseThrow(() -> new RuntimeException("Failed to schedule job " + job.getId())))
+                .whenCompleteAsync((r, t) -> Optional
+                        .ofNullable(t)
+                        .ifPresent(ex -> LOGGER.error("Error Scheduling Job: {}. Details: {}", job, t)));
+        return response;
+    }
+
+    @DELETE
+    @Produces(MediaType.APPLICATION_JSON)
+    @Path("/{id}")
+    public CompletionStage<Job> delete(@PathParam("id") String id) {
+        LOGGER.debug("REST delete id {}", id);
+        return  scheduler
+                .cancel(id)
+                .thenApply(result -> Optional
+                        .ofNullable(result)
+                        .map(ScheduledJob::getJob)
+                        .orElseThrow(() -> new RuntimeException("Failed to cancel job scheduling")));
+    }
+
+    @GET
+    @Produces(MediaType.APPLICATION_JSON)
+    @Path("/{id}")
+    public CompletionStage<Job> get(@PathParam("id") String id) {
+        LOGGER.debug("REST get {}", id);
+        return reactiveJobRepository
+                .get(id)
+                .thenApply(ScheduledJob::getJob);
+    }
+}
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/error/DefaultErrorMapper.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/error/DefaultErrorMapper.java
new file mode 100644
index 00000000000..42b52908511
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/resource/error/DefaultErrorMapper.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.resource.error;
+
+import java.util.Optional;
+
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.ext.ExceptionMapper;
+import javax.ws.rs.ext.Provider;
+
+import org.kie.kogito.jobs.service.model.ErrorResponse;
+
+@Provider
+public class DefaultErrorMapper implements ExceptionMapper<Exception> {
+
+    @Override
+    public Response toResponse(Exception exception) {
+        return Response.status(Optional.ofNullable(exception)
+                                       .filter(WebApplicationException.class::isInstance)
+                                       .map(WebApplicationException.class::cast)
+                                       .map(WebApplicationException::getResponse)
+                                       .map(Response::getStatus)
+                                       .orElse(500))
+                .type(MediaType.APPLICATION_JSON_TYPE)
+                .entity(new ErrorResponse(exception.getMessage()))
+                .build();
+    }
+}
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/BaseTimerJobScheduler.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/BaseTimerJobScheduler.java
new file mode 100644
index 00000000000..386242e3870
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/BaseTimerJobScheduler.java
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.scheduler;
+
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.TemporalUnit;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import javax.inject.Inject;
+
+import org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder;
+import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
+import org.kie.kogito.jobs.api.Job;
+import org.kie.kogito.jobs.service.executor.JobExecutor;
+import org.kie.kogito.jobs.service.model.ScheduledJob;
+import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
+import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base reactive Job Scheduler that performs the fundamental operations and let to the concrete classes to
+ * implement the scheduling actions.
+ */
+public abstract class BaseTimerJobScheduler implements ReactiveJobScheduler<ScheduledJob> {
+
+    private Logger logger = LoggerFactory.getLogger(BaseTimerJobScheduler.class);
+
+    @Inject
+    private JobExecutor jobExecutor;
+
+    @Inject
+    private ReactiveJobRepository jobRepository;
+
+    @Override
+    public Publisher<ScheduledJob> schedule(Job job) {
+        return ReactiveStreams
+                //1- check if the job is already scheduled
+                .fromCompletionStage(jobRepository.exists(job.getId()))
+                .flatMapCompletionStage(exists -> exists
+                        ? cancel(job.getId()).thenApply(Objects::nonNull)
+                        : CompletableFuture.completedFuture(Boolean.TRUE))
+                .filter(Boolean.TRUE::equals)
+                //2- calculate the delay (when the job should be executed)
+                .map(checked -> job.getExpirationTime())
+                .map(expirationTime -> Duration.between(ZonedDateTime.now(ZoneId.of("UTC")), expirationTime))
+                //3- schedule the job
+                .map(delay -> doSchedule(delay, job))
+                .flatMapRsPublisher(p -> p)
+                .map(scheduledJob -> jobRepository.save(scheduledJob))
+                .flatMapCompletionStage(p -> p)
+                .buildRs();
+    }
+
+    public abstract Publisher<ScheduledJob> doSchedule(Duration delay, Job job);
+
+    protected CompletionStage<Job> execute(Job job) {
+        logger.info("Job executed ! {}", job);
+        return jobExecutor.execute(job);
+    }
+
+    @Override
+    public CompletionStage<ScheduledJob> cancel(String jobId) {
+        logger.debug("Cancel Job Scheduling {}", jobId);
+        return ReactiveStreams
+                .fromCompletionStageNullable(jobRepository.get(jobId))
+                .flatMapRsPublisher(this::doCancel)
+                .filter(Boolean.TRUE::equals)
+                .map(r -> jobRepository.delete(jobId))
+                .findFirst()
+                .run()
+                .thenCompose(job -> job.orElseThrow(()-> new RuntimeException("Failed to cancel job scheduling " + jobId)));
+    }
+
+    public abstract Publisher<Boolean> doCancel(ScheduledJob scheduledJob);
+}
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/JobScheduler.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/JobScheduler.java
new file mode 100644
index 00000000000..838b3226c63
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/JobScheduler.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.scheduler;
+
+import org.kie.kogito.jobs.api.Job;
+
+public interface JobScheduler<T, C> {
+
+    T schedule(Job job);
+
+    C cancel(String jobId);
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/ReactiveJobScheduler.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/ReactiveJobScheduler.java
new file mode 100644
index 00000000000..63948fc2930
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/ReactiveJobScheduler.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.scheduler;
+
+import java.util.concurrent.CompletionStage;
+
+import org.kie.kogito.jobs.api.Job;
+import org.kie.kogito.jobs.service.model.ScheduledJob;
+import org.reactivestreams.Publisher;
+
+public interface ReactiveJobScheduler<T> extends JobScheduler<Publisher<T>, CompletionStage<ScheduledJob>> {
+
+    Publisher<T> schedule(Job job);
+
+    CompletionStage<ScheduledJob> cancel(String jobId);
+}
diff --git a/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/impl/VertxJobScheduler.java b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/impl/VertxJobScheduler.java
new file mode 100644
index 00000000000..eb6669bd705
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/java/org/kie/kogito/jobs/service/scheduler/impl/VertxJobScheduler.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.scheduler.impl;
+
+import java.time.Duration;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+import io.vertx.axle.core.Vertx;
+import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
+import org.kie.kogito.jobs.api.Job;
+import org.kie.kogito.jobs.service.model.ScheduledJob;
+import org.kie.kogito.jobs.service.scheduler.BaseTimerJobScheduler;
+import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Job Scheduler based on Vert.x engine.
+ */
+@ApplicationScoped
+public class VertxJobScheduler extends BaseTimerJobScheduler {
+
+    private Logger logger = LoggerFactory.getLogger(VertxJobScheduler.class);
+
+    @Inject
+    private Vertx vertx;
+
+    @Override
+    public Publisher<ScheduledJob> doSchedule(Duration delay, Job job) {
+        logger.debug("Job Scheduling {}", job);
+        return ReactiveStreams
+                .of(setTimer(delay, job))
+                .map(id -> new ScheduledJob(job, String.valueOf(id)))
+                .buildRs();
+    }
+
+    private long setTimer(Duration delay, Job job) {
+        return vertx.setTimer(delay.toMillis(), scheduledId -> execute(job));
+    }
+
+    @Override
+    public Publisher<Boolean> doCancel(ScheduledJob scheduledJob) {
+        return ReactiveStreams
+                .of(scheduledJob)
+                .map(ScheduledJob::getScheduledId)
+                .map(Long::valueOf)
+                .map(vertx::cancelTimer)
+                .buildRs();
+    }
+}
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/main/resources/application.properties b/addons/jobs/jobs-service/src/main/resources/application.properties
new file mode 100644
index 00000000000..dd004efa9eb
--- /dev/null
+++ b/addons/jobs/jobs-service/src/main/resources/application.properties
@@ -0,0 +1,44 @@
+#
+# Copyright 2019 Red Hat, Inc. and/or its affiliates.
+#
+# Licensed 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.
+#
+
+#Log Config
+quarkus.log.level=ERROR
+quarkus.log.category."org.kie.kogito.jobs".level=DEBUG
+
+##Console
+quarkus.log.console.enable=true
+quarkus.log.console.format=%d{yyyy-MM-dd HH:mm:ss,SSS} %h %-5p [%c] (%t) %s%e%n
+quarkus.log.console.color=true
+quarkus.log.console.async=false
+##File
+quarkus.log.file.enable=true
+quarkus.log.file.format=%d{yyyy-MM-dd HH:mm:ss,SSS} %h %-5p [%c] (%t) %s%e%n
+quarkus.log.file.path=jobs-service.log
+quarkus.log.file.async=true
+quarkus.log.file.rotation.file-suffix=yyyy-MM-dd
+
+#Web Config
+quarkus.ssl.native=true
+quarkus.resteasy.gzip.enabled=true 
+quarkus.resteasy.gzip.max-input=10M 
+quarkus.http.cors=true
+quarkus.http.cors.methods=GET,PUT,POST,DELETE
+quarkus.http.limits.max-body-size=10M
+
+#Swagger
+quarkus.smallrye-openapi.path=/swagger
+quarkus.swagger-ui.always-include=true
+quarkus.swagger-ui.path=/swagger-ui
\ No newline at end of file
diff --git a/addons/jobs/jobs-service/src/test/java/org/kie/kogito/jobs/service/resource/JobResourceTest.java b/addons/jobs/jobs-service/src/test/java/org/kie/kogito/jobs/service/resource/JobResourceTest.java
new file mode 100644
index 00000000000..6d08bb5c631
--- /dev/null
+++ b/addons/jobs/jobs-service/src/test/java/org/kie/kogito/jobs/service/resource/JobResourceTest.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright 2019 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed 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.kie.kogito.jobs.service.resource;
+
+import java.time.ZonedDateTime;
+
+import javax.inject.Inject;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.http.ContentType;
+import io.restassured.response.ValidatableResponse;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.kie.kogito.jobs.api.Job;
+import org.kie.kogito.jobs.api.JobBuilder;
+import org.kie.kogito.jobs.service.json.JacksonConfiguration;
+
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.CoreMatchers.equalTo;
+
+@QuarkusTest
+class JobResourceTest {
+
+    @Inject
+    private JacksonConfiguration jacksonConfiguration;
+
+    @BeforeEach
+    void setUp() {
+    }
+
+    @Test
+    void create() throws Exception {
+        final String body = getJob("1");
+        create(body);
+    }
+
+    private ValidatableResponse create(String body) {
+        return given()
+                .contentType(ContentType.JSON)
+                .body(body)
+                .when()
+                .post("/job")
+                .then()
+                .statusCode(200)
+                .contentType(ContentType.JSON)
+                .assertThat()
+                .body(equalTo(body))
+                .log()
+                .body();
+    }
+
+    private String getJob(String id) throws JsonProcessingException {
+        final Job job = JobBuilder
+                .builder()
+                .id(id)
+                .expirationTime(ZonedDateTime.now().plusMinutes(1))
+                .priority(1)
+                .build();
+        return jacksonConfiguration.objectMapper().writeValueAsString(job);
+    }
+
+    @Test
+    void deleteAfterCreate() throws Exception {
+        final String id = "2";
+        final String body = getJob(id);
+        create(body);
+        given().pathParam("id", id)
+                .when()
+                .delete("/job/{id}")
+                .then()
+                .statusCode(200)
+                .contentType(ContentType.JSON)
+                .assertThat()
+                .body(equalTo(body))
+                .log()
+                .body();
+    }
+
+    @Test
+    void getAfterCreate() throws Exception {
+        final String id = "3";
+        final String body = getJob(id);
+        create(body);
+        given()
+                .pathParam("id", id)
+                .when()
+                .get("/job/{id}")
+                .then()
+                .statusCode(200)
+                .contentType(ContentType.JSON)
+                .assertThat()
+                .body(equalTo(body))
+                .log()
+                .body();
+    }
+}
\ No newline at end of file