You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by ro...@apache.org on 2019/09/03 07:34:22 UTC
[james-project] branch master updated (c7125d4 -> fda3446)
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from c7125d4 MAILBOX-359 Remove no longer used classes
new 292e6dc JAMES-2813 Use RabbitMQWorkqueue in the distributed TaskManager test
new 5ed834b JAMES-2813 Implement TaskManagerContract for DistributedTaskManager
new f6b8deb JAMES-2813 Add DistributedTaskManagerTest cases with multiple TaskManagers
new aaf5257 JAMES-2813 Make RabbitMQWorkQueue cluster-serialized
new 1c8295a JAMES-2873 Add test to ensure DistributedTaskManager track nodes correctly
new fda3446 JAMES-2813 Fix distributed task manager runtime dependency
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/james/DistributedTaskManagerModule.java | 12 +-
server/task-distributed/pom.xml | 20 +++
.../distributed/RabbitMQExclusiveConsumer.java | 174 +++++++++++++++++++++
.../distributed/RabbitMQWorkQueue.java | 146 +++++++++++++++++
.../distributed/RabbitMQWorkQueueSupplier.scala} | 21 ++-
.../distributed/DistributedTaskManagerTest.java | 138 ++++++++++++----
.../distributed/RabbitMQWorkQueueTest.java | 147 +++++++++++++++++
.../apache/james/task/SerialTaskManagerWorker.java | 15 +-
.../org/apache/james/task/TaskManagerWorker.java | 2 +
9 files changed, 627 insertions(+), 48 deletions(-)
create mode 100644 server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQExclusiveConsumer.java
create mode 100644 server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
copy server/{task/src/main/scala/org/apache/james/task/eventsourcing/WorkQueueSupplier.scala => task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueSupplier.scala} (55%)
create mode 100644 server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/06: JAMES-2813 Make RabbitMQWorkQueue
cluster-serialized
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit aaf5257c40abac268fe1d1f0cb344cf7b9023963
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Thu Aug 22 16:38:40 2019 +0200
JAMES-2813 Make RabbitMQWorkQueue cluster-serialized
---
server/task-distributed/pom.xml | 10 ++
.../distributed/RabbitMQExclusiveConsumer.java | 174 +++++++++++++++++++++
.../distributed/RabbitMQWorkQueue.java | 146 +++++++++++++++++
.../distributed/RabbitMQWorkQueueTest.java | 147 +++++++++++++++++
.../apache/james/task/SerialTaskManagerWorker.java | 15 +-
.../org/apache/james/task/TaskManagerWorker.java | 2 +
6 files changed, 489 insertions(+), 5 deletions(-)
diff --git a/server/task-distributed/pom.xml b/server/task-distributed/pom.xml
index 844c693..89b4edf 100644
--- a/server/task-distributed/pom.xml
+++ b/server/task-distributed/pom.xml
@@ -39,6 +39,16 @@
</dependency>
<dependency>
<groupId>${james.groupId}</groupId>
+ <artifactId>apache-james-backends-rabbitmq</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>apache-james-backends-rabbitmq</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
<artifactId>apache-james-backends-cassandra</artifactId>
<type>test-jar</type>
<scope>test</scope>
diff --git a/server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQExclusiveConsumer.java b/server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQExclusiveConsumer.java
new file mode 100644
index 0000000..2977312
--- /dev/null
+++ b/server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQExclusiveConsumer.java
@@ -0,0 +1,174 @@
+/****************************************************************
+ * 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.james.task.eventsourcing.distributed;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.ImmutableMap;
+import com.rabbitmq.client.CancelCallback;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.DeliverCallback;
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+import reactor.rabbitmq.AcknowledgableDelivery;
+import reactor.rabbitmq.ConsumeOptions;
+import reactor.rabbitmq.RabbitFluxException;
+import reactor.rabbitmq.ReceiverOptions;
+
+/**
+ * Taken from {@link reactor.rabbitmq.Receiver}
+ * In order to be able to set the `exclusive` parameter to `true`
+ * to the `channel.basicConsume` method.
+ *
+ * @deprecated to remove once the parallel execution of task has been implemented
+ */
+@Deprecated
+public class RabbitMQExclusiveConsumer implements Closeable {
+ private static final Function<Connection, Channel> CHANNEL_CREATION_FUNCTION = new RabbitMQExclusiveConsumer.ChannelCreationFunction();
+ private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQExclusiveConsumer.class);
+
+ private static final boolean NON_LOCAL = true;
+ private static final boolean EXCLUSIVE = true;
+
+ private static class ChannelCreationFunction implements Function<Connection, Channel> {
+
+ @Override
+ public Channel apply(Connection connection) {
+ try {
+ return connection.createChannel();
+ } catch (IOException e) {
+ throw new RabbitFluxException("Error while creating channel", e);
+ }
+ }
+ }
+
+ private Mono<? extends Connection> connectionMono;
+ private final AtomicBoolean hasConnection;
+ private final Scheduler connectionSubscriptionScheduler;
+ private final boolean privateConnectionSubscriptionScheduler;
+
+ public RabbitMQExclusiveConsumer(ReceiverOptions options) {
+ this.privateConnectionSubscriptionScheduler = options.getConnectionSubscriptionScheduler() == null;
+ this.connectionSubscriptionScheduler = options.getConnectionSubscriptionScheduler() == null ?
+ createScheduler("rabbitmq-receiver-connection-subscription") : options.getConnectionSubscriptionScheduler();
+ hasConnection = new AtomicBoolean(false);
+ this.connectionMono = options.getConnectionMono() != null ? options.getConnectionMono() :
+ Mono.fromCallable(() -> options.getConnectionFactory().newConnection())
+ .doOnSubscribe(c -> hasConnection.set(true))
+ .subscribeOn(this.connectionSubscriptionScheduler)
+ .cache();
+ }
+
+ protected Scheduler createScheduler(String name) {
+ return Schedulers.newElastic(name);
+ }
+
+
+ public Flux<AcknowledgableDelivery> consumeExclusiveManualAck(final String queue, ConsumeOptions options) {
+ // TODO track flux so it can be disposed when the sender is closed?
+ // could be also developer responsibility
+ return Flux.create(emitter -> connectionMono.map(CHANNEL_CREATION_FUNCTION).subscribe(channel -> {
+ try {
+ if (options.getQos() != 0) {
+ channel.basicQos(options.getQos());
+ }
+
+ DeliverCallback deliverCallback = (consumerTag, message) -> {
+ AcknowledgableDelivery delivery = new AcknowledgableDelivery(message, channel, options.getExceptionHandler());
+ if (options.getHookBeforeEmitBiFunction().apply(emitter.requestedFromDownstream(), delivery)) {
+ emitter.next(delivery);
+ }
+ if (options.getStopConsumingBiFunction().apply(emitter.requestedFromDownstream(), message)) {
+ emitter.complete();
+ }
+ };
+
+ AtomicBoolean basicCancel = new AtomicBoolean(true);
+ CancelCallback cancelCallback = consumerTag -> {
+ LOGGER.info("Flux consumer {} has been cancelled", consumerTag);
+ basicCancel.set(false);
+ emitter.complete();
+ };
+
+ completeOnChannelShutdown(channel, emitter);
+
+ Map<String, Object> arguments = ImmutableMap.of();
+ final String consumerTag = channel.basicConsume(queue, false, UUID.randomUUID().toString(), !NON_LOCAL, EXCLUSIVE, arguments, deliverCallback, cancelCallback);
+ AtomicBoolean cancelled = new AtomicBoolean(false);
+ LOGGER.info("Consumer {} consuming from {} has been registered", consumerTag, queue);
+ emitter.onDispose(() -> {
+ LOGGER.info("Cancelling consumer {} consuming from {}", consumerTag, queue);
+ if (cancelled.compareAndSet(false, true)) {
+ try {
+ if (channel.isOpen() && channel.getConnection().isOpen()) {
+ if (basicCancel.compareAndSet(true, false)) {
+ channel.basicCancel(consumerTag);
+ }
+ channel.close();
+ }
+ } catch (TimeoutException | IOException e) {
+ // Not sure what to do, not much we can do,
+ // logging should be enough.
+ // Maybe one good reason to introduce an exception handler to choose more easily.
+ LOGGER.warn("Error while closing channel: " + e.getMessage());
+ }
+ }
+ });
+ } catch (IOException e) {
+ throw new RabbitFluxException(e);
+ }
+ }, emitter::error), options.getOverflowStrategy());
+ }
+
+ protected void completeOnChannelShutdown(Channel channel, FluxSink<?> emitter) {
+ channel.addShutdownListener(reason -> {
+ if (!AutorecoveringConnection.DEFAULT_CONNECTION_RECOVERY_TRIGGERING_CONDITION.test(reason)) {
+ emitter.complete();
+ }
+ });
+ }
+
+ public void close() {
+ if (hasConnection.getAndSet(false)) {
+ try {
+ // FIXME use timeout on block (should be a parameter of the Receiver)
+ connectionMono.block().close();
+ } catch (IOException e) {
+ throw new RabbitFluxException(e);
+ }
+ }
+ if (privateConnectionSubscriptionScheduler) {
+ this.connectionSubscriptionScheduler.dispose();
+ }
+ }
+}
diff --git a/server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java b/server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
new file mode 100644
index 0000000..69cdf16
--- /dev/null
+++ b/server/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
@@ -0,0 +1,146 @@
+/**
+ * *************************************************************
+ * 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.james.task.eventsourcing.distributed;
+
+import java.nio.charset.StandardCharsets;
+
+import javax.annotation.PreDestroy;
+
+import org.apache.commons.lang3.NotImplementedException;
+import org.apache.james.backend.rabbitmq.ReactorRabbitMQChannelPool;
+import org.apache.james.backend.rabbitmq.SimpleConnectionPool;
+import org.apache.james.lifecycle.api.Startable;
+import org.apache.james.server.task.json.JsonTaskSerializer;
+import org.apache.james.task.Task;
+import org.apache.james.task.TaskId;
+import org.apache.james.task.TaskManagerWorker;
+import org.apache.james.task.TaskWithId;
+import org.apache.james.task.WorkQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.github.fge.lambdas.Throwing;
+import com.google.common.collect.ImmutableMap;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Connection;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import reactor.rabbitmq.AcknowledgableDelivery;
+import reactor.rabbitmq.BindingSpecification;
+import reactor.rabbitmq.ConsumeOptions;
+import reactor.rabbitmq.ExchangeSpecification;
+import reactor.rabbitmq.OutboundMessage;
+import reactor.rabbitmq.QueueSpecification;
+import reactor.rabbitmq.RabbitFlux;
+import reactor.rabbitmq.ReceiverOptions;
+import reactor.rabbitmq.Sender;
+import reactor.rabbitmq.SenderOptions;
+
+public class RabbitMQWorkQueue implements WorkQueue, Startable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQWorkQueue.class);
+
+ static final Integer MAX_CHANNELS_NUMBER = 1;
+ static final String EXCHANGE_NAME = "taskManagerWorkQueueExchange";
+ static final String QUEUE_NAME = "taskManagerWorkQueue";
+ static final String ROUTING_KEY = "taskManagerWorkQueueRoutingKey";
+ public static final String TASK_ID = "taskId";
+
+ private final TaskManagerWorker worker;
+ private final Mono<Connection> connectionMono;
+ private final ReactorRabbitMQChannelPool channelPool;
+ private final JsonTaskSerializer taskSerializer;
+ private Sender sender;
+ private RabbitMQExclusiveConsumer receiver;
+
+ public RabbitMQWorkQueue(TaskManagerWorker worker, SimpleConnectionPool simpleConnectionPool, JsonTaskSerializer taskSerializer) {
+ this.worker = worker;
+ this.connectionMono = simpleConnectionPool.getResilientConnection();
+ this.taskSerializer = taskSerializer;
+ this.channelPool = new ReactorRabbitMQChannelPool(connectionMono, MAX_CHANNELS_NUMBER);
+ }
+
+ public void start() {
+ sender = RabbitFlux.createSender(new SenderOptions()
+ .connectionMono(connectionMono)
+ .channelPool(channelPool)
+ .resourceManagementChannelMono(
+ connectionMono.map(Throwing.function(Connection::createChannel)).cache()));
+
+ receiver = new RabbitMQExclusiveConsumer(new ReceiverOptions().connectionMono(connectionMono));
+
+ sender.declareExchange(ExchangeSpecification.exchange(EXCHANGE_NAME)).block();
+ sender.declare(QueueSpecification.queue(QUEUE_NAME).durable(true)).block();
+ sender.bind(BindingSpecification.binding(EXCHANGE_NAME, ROUTING_KEY, QUEUE_NAME)).block();
+
+ consumeWorkqueue();
+ }
+
+ private void consumeWorkqueue() {
+ receiver.consumeExclusiveManualAck(QUEUE_NAME, new ConsumeOptions())
+ .subscribeOn(Schedulers.elastic())
+ .flatMap(this::executeTask)
+ .subscribe();
+ }
+
+ private Mono<Task.Result> executeTask(AcknowledgableDelivery delivery) {
+ String json = new String(delivery.getBody(), StandardCharsets.UTF_8);
+
+ TaskId taskId = TaskId.fromString(delivery.getProperties().getHeaders().get(TASK_ID).toString());
+
+ try {
+ Task task = taskSerializer.deserialize(json);
+ delivery.ack();
+ return worker.executeTask(new TaskWithId(taskId, task));
+ } catch (Exception e) {
+ LOGGER.error("Unable to run submitted Task " + taskId.asString(), e);
+ delivery.ack();
+ worker.fail(taskId, e);
+ return Mono.empty();
+ }
+ }
+
+ @Override
+ public void submit(TaskWithId taskWithId) {
+ try {
+ byte[] payload = taskSerializer.serialize(taskWithId.getTask()).getBytes(StandardCharsets.UTF_8);
+ AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder()
+ .headers(ImmutableMap.of(TASK_ID, taskWithId.getId().asString()))
+ .build();
+ OutboundMessage outboundMessage = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload);
+ sender.send(Mono.just(outboundMessage)).block();
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void cancel(TaskId taskId) {
+ throw new NotImplementedException("Cancel not done yet");
+ }
+
+ @Override
+ @PreDestroy
+ public void close() {
+ channelPool.close();
+ }
+}
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java
new file mode 100644
index 0000000..2383695
--- /dev/null
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java
@@ -0,0 +1,147 @@
+/**
+ * *************************************************************
+ * 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.james.task.eventsourcing.distributed;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.stream.IntStream;
+
+import org.apache.james.backend.rabbitmq.RabbitMQExtension;
+import org.apache.james.server.task.json.JsonTaskSerializer;
+import org.apache.james.server.task.json.TestTask;
+import org.apache.james.server.task.json.dto.TestTaskDTOModules;
+import org.apache.james.task.CompletedTask;
+import org.apache.james.task.Task;
+import org.apache.james.task.TaskId;
+import org.apache.james.task.TaskManagerWorker;
+import org.apache.james.task.TaskWithId;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.mockito.ArgumentCaptor;
+import org.mockito.verification.Timeout;
+
+import reactor.core.publisher.Mono;
+
+class RabbitMQWorkQueueTest {
+ private static final TaskId TASK_ID = TaskId.fromString("2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd");
+ private static final TaskId TASK_ID_2 = TaskId.fromString("3c7f4081-aa30-11e9-bf6c-2d3b9e84aafd");
+ private static final Task TASK = new CompletedTask();
+ private static final Task TASK2 = new CompletedTask();
+ private static final TaskWithId TASK_WITH_ID = new TaskWithId(TASK_ID, TASK);
+ private static final TaskWithId TASK_WITH_ID_2 = new TaskWithId(TASK_ID_2, TASK2);
+
+ @RegisterExtension
+ static RabbitMQExtension rabbitMQExtension = RabbitMQExtension.singletonRabbitMQ();
+
+ private RabbitMQWorkQueue testee;
+ private TaskManagerWorker taskManagerWorker;
+ private JsonTaskSerializer taskSerializer;
+
+ @BeforeEach
+ void setUp() {
+ taskManagerWorker = mock(TaskManagerWorker.class);
+ when(taskManagerWorker.executeTask(TASK_WITH_ID)).thenReturn(Mono.just(Task.Result.COMPLETED));
+ when(taskManagerWorker.executeTask(TASK_WITH_ID_2)).thenReturn(Mono.just(Task.Result.COMPLETED));
+ taskSerializer = new JsonTaskSerializer(TestTaskDTOModules.COMPLETED_TASK_MODULE);
+ testee = new RabbitMQWorkQueue(taskManagerWorker, rabbitMQExtension.getRabbitConnectionPool(), taskSerializer);
+ testee.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ testee.close();
+ }
+
+ @Test
+ void workerShouldConsumeSubmittedTask() {
+ testee.submit(TASK_WITH_ID);
+
+ ArgumentCaptor<TaskWithId> taskWithIdCaptor = ArgumentCaptor.forClass(TaskWithId.class);
+ verify(taskManagerWorker, timeout(1000)).executeTask(taskWithIdCaptor.capture());
+
+ TaskWithId actualTaskWithId = taskWithIdCaptor.getValue();
+ assertThat(actualTaskWithId.getId()).isEqualTo(TASK_ID);
+ assertThat(actualTaskWithId.getTask().type()).isEqualTo(TASK.type());
+ }
+
+ @Test
+ void workerShouldConsumeTwoSubmittedTask() {
+ testee.submit(TASK_WITH_ID);
+ testee.submit(TASK_WITH_ID_2);
+
+ ArgumentCaptor<TaskWithId> taskWithIdCaptor = ArgumentCaptor.forClass(TaskWithId.class);
+ verify(taskManagerWorker, new Timeout(1000, times(2))).executeTask(taskWithIdCaptor.capture());
+
+ TaskWithId actualTaskWithId = taskWithIdCaptor.getAllValues().get(0);
+ assertThat(actualTaskWithId.getId()).isEqualTo(TASK_ID);
+ assertThat(actualTaskWithId.getTask().type()).isEqualTo(TASK.type());
+
+ TaskWithId actualSecondTaskWithId = taskWithIdCaptor.getAllValues().get(1);
+ assertThat(actualSecondTaskWithId.getId()).isEqualTo(TASK_ID_2);
+ assertThat(actualSecondTaskWithId.getTask().type()).isEqualTo(TASK2.type());
+ }
+
+ @Test
+ void givenTwoWorkQueuesOnlyTheFirstOneIsConsumingTasks() {
+ testee.submit(TASK_WITH_ID);
+
+ TaskManagerWorker otherTaskManagerWorker = mock(TaskManagerWorker.class);
+ RabbitMQWorkQueue otherWorkQueue = new RabbitMQWorkQueue(otherTaskManagerWorker, rabbitMQExtension.getRabbitConnectionPool(), taskSerializer);
+ otherWorkQueue.start();
+
+ IntStream.range(0, 9)
+ .forEach(ignoredIndex -> testee.submit(TASK_WITH_ID_2));
+
+ verify(taskManagerWorker, new Timeout(1000, times(10))).executeTask(any());
+
+ verify(otherTaskManagerWorker, new Timeout(1000, times(0))).executeTask(any());
+ }
+
+ @Test
+ void givenANonDeserializableTaskItShouldBeFlaggedAsFailedAndItDoesNotPreventFollowingTasks() throws InterruptedException {
+ Task task = new TestTask(42);
+ TaskId taskId = TaskId.fromString("4bf6d081-aa30-11e9-bf6c-2d3b9e84aafd");
+ TaskWithId taskWithId = new TaskWithId(taskId, task);
+
+ TaskManagerWorker otherTaskManagerWorker = mock(TaskManagerWorker.class);
+ JsonTaskSerializer otherTaskSerializer = new JsonTaskSerializer(TestTaskDTOModules.TEST_TYPE);
+ RabbitMQWorkQueue otherWorkQueue = new RabbitMQWorkQueue(otherTaskManagerWorker, rabbitMQExtension.getRabbitConnectionPool(), otherTaskSerializer);
+ //wait to be sur that the first workqueue has subscribed as an exclusive consumer of the RabbitMQ queue.
+ Thread.sleep(200);
+ otherWorkQueue.start();
+
+ otherWorkQueue.submit(taskWithId);
+
+ verify(taskManagerWorker, new Timeout(100, times(0))).executeTask(any());
+ verify(taskManagerWorker, timeout(100)).fail(eq(taskId), any());
+
+ testee.submit(TASK_WITH_ID);
+ verify(taskManagerWorker, timeout(100)).executeTask(any());
+ }
+}
diff --git a/server/task/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java b/server/task/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java
index afb7a39..c5a0638 100644
--- a/server/task/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java
+++ b/server/task/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java
@@ -60,11 +60,11 @@ public class SerialTaskManagerWorker implements TaskManagerWorker {
@Override
public Mono<Task.Result> executeTask(TaskWithId taskWithId) {
- return Mono
- .using(
- acquireSemaphore(taskWithId, listener),
- executeWithSemaphore(taskWithId, listener),
- Semaphore::release);
+ return Mono
+ .using(
+ acquireSemaphore(taskWithId, listener),
+ executeWithSemaphore(taskWithId, listener),
+ Semaphore::release);
}
@@ -140,6 +140,11 @@ public class SerialTaskManagerWorker implements TaskManagerWorker {
}
@Override
+ public void fail(TaskId taskId, Throwable reason) {
+ listener.failed(taskId, reason);
+ }
+
+ @Override
public void close() throws IOException {
taskExecutor.shutdownNow();
}
diff --git a/server/task/src/main/java/org/apache/james/task/TaskManagerWorker.java b/server/task/src/main/java/org/apache/james/task/TaskManagerWorker.java
index 4780cc7..94051c7 100644
--- a/server/task/src/main/java/org/apache/james/task/TaskManagerWorker.java
+++ b/server/task/src/main/java/org/apache/james/task/TaskManagerWorker.java
@@ -39,4 +39,6 @@ public interface TaskManagerWorker extends Closeable {
Mono<Task.Result> executeTask(TaskWithId taskWithId);
void cancelTask(TaskId taskId);
+
+ void fail(TaskId taskId, Throwable reason);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/06: JAMES-2813 Add DistributedTaskManagerTest
cases with multiple TaskManagers
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit f6b8deb6d0b7daeb2e4614bf7dda0b2d2dea3ef2
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Thu Aug 22 15:07:56 2019 +0200
JAMES-2813 Add DistributedTaskManagerTest cases with multiple TaskManagers
---
.../distributed/DistributedTaskManagerTest.java | 43 +++++++++++++++++++++-
1 file changed, 42 insertions(+), 1 deletion(-)
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
index 7561181..994edca 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
@@ -40,6 +40,8 @@ import org.apache.james.server.task.json.dto.MemoryReferenceTaskStore;
import org.apache.james.server.task.json.dto.TestTaskDTOModules;
import org.apache.james.task.CompletedTask;
import org.apache.james.task.CountDownLatchExtension;
+import org.apache.james.task.MemoryReferenceTask;
+import org.apache.james.task.Task;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskId;
import org.apache.james.task.TaskManager;
@@ -91,11 +93,12 @@ class DistributedTaskManagerTest implements TaskManagerContract {
private final CassandraTaskExecutionDetailsProjectionDAO cassandraTaskExecutionDetailsProjectionDAO = new CassandraTaskExecutionDetailsProjectionDAO(cassandra.getConf(), cassandra.getTypesProvider());
private final TaskExecutionDetailsProjection executionDetailsProjection = new CassandraTaskExecutionDetailsProjection(cassandraTaskExecutionDetailsProjectionDAO);
- WorkQueueSupplier workQueueSupplier = new RabbitMQWorkQueueSupplier(rabbitMQExtension.getRabbitConnectionPool(), TASK_SERIALIZER);
+ private WorkQueueSupplier workQueueSupplier;
private EventStore eventStore;
@BeforeEach
void setUp(EventStore eventStore) {
+ workQueueSupplier = new RabbitMQWorkQueueSupplier(rabbitMQExtension.getRabbitConnectionPool(), TASK_SERIALIZER);
this.eventStore = eventStore;
}
@@ -128,4 +131,42 @@ class DistributedTaskManagerTest implements TaskManagerContract {
@Disabled("Cancelling is not supported yet")
public void getStatusShouldBeCancelledWhenCancelled(CountDownLatch countDownLatch) {
}
+
+ @Test
+ void givenTwoTaskManagersAndTwoTasksOnlyOneTaskShouldRunAtTheSameTime() throws InterruptedException {
+ CountDownLatch waitingForFirstTaskLatch = new CountDownLatch(1);
+
+ TaskManager taskManager1 = taskManager();
+ TaskManager taskManager2 = taskManager();
+
+ taskManager1.submit(new MemoryReferenceTask(() -> {
+ waitingForFirstTaskLatch.await();
+ return Task.Result.COMPLETED;
+ }));
+ TaskId waitingTaskId = taskManager1.submit(new CompletedTask());
+
+ awaitUntilTaskHasStatus(waitingTaskId, TaskManager.Status.WAITING, taskManager2);
+ waitingForFirstTaskLatch.countDown();
+
+ Awaitility.await()
+ .atMost(Duration.ONE_SECOND)
+ .pollInterval(100L, TimeUnit.MILLISECONDS)
+ .until(() -> taskManager1.await(waitingTaskId).getStatus() == TaskManager.Status.COMPLETED);
+ }
+
+ @Test
+ // FIXME it's currently dependent of the implementation of the sequential TaskManager with the exclusive RabbitMQ consumer
+ // once we store the node where the event have been created/started/completed we should rewrite it with this information.
+ void givenTwoTaskManagerATaskSubmittedOnOneCouldBeRunOnTheOther() throws InterruptedException {
+ TaskManager taskManager1 = taskManager();
+ Thread.sleep(100); // FIXME used to ensure that taskManager1 is the worker consuming from rabbit
+ TaskManager taskManager2 = taskManager();
+
+ TaskId taskId = taskManager2.submit(new CompletedTask());
+
+ Awaitility.await()
+ .atMost(Duration.ONE_SECOND)
+ .pollInterval(100L, TimeUnit.MILLISECONDS)
+ .until(() -> taskManager1.await(taskId).getStatus() == TaskManager.Status.COMPLETED);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 05/06: JAMES-2873 Add test to ensure
DistributedTaskManager track nodes correctly
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 1c8295a687ca95169ab67916f83503018498cad6
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Mon Aug 26 11:32:34 2019 +0200
JAMES-2873 Add test to ensure DistributedTaskManager track nodes correctly
---
.../distributed/DistributedTaskManagerTest.java | 87 ++++++++++++----------
1 file changed, 48 insertions(+), 39 deletions(-)
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
index 994edca..ca70292 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
@@ -102,24 +102,28 @@ class DistributedTaskManagerTest implements TaskManagerContract {
this.eventStore = eventStore;
}
- public TaskManager taskManager() {
+ public EventSourcingTaskManager taskManager() {
return new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME);
}
+ private EventSourcingTaskManager taskManager(Hostname hostname) {
+ return new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, hostname);
+ }
+
@Test
void givenOneEventStoreTwoEventTaskManagersShareTheSameEvents() {
- TaskManager taskManager1 = taskManager();
- TaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME_2);
-
- TaskId taskId = taskManager1.submit(new CompletedTask());
- Awaitility.await()
- .atMost(Duration.FIVE_SECONDS)
- .pollInterval(100L, TimeUnit.MILLISECONDS)
- .until(() -> taskManager1.await(taskId).getStatus() == TaskManager.Status.COMPLETED);
-
- TaskExecutionDetails detailsFromTaskManager1 = taskManager1.getExecutionDetails(taskId);
- TaskExecutionDetails detailsFromTaskManager2 = taskManager2.getExecutionDetails(taskId);
- assertThat(detailsFromTaskManager1).isEqualTo(detailsFromTaskManager2);
+ try (EventSourcingTaskManager taskManager1 = taskManager();
+ EventSourcingTaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME_2)) {
+ TaskId taskId = taskManager1.submit(new CompletedTask());
+ Awaitility.await()
+ .atMost(Duration.FIVE_SECONDS)
+ .pollInterval(100L, TimeUnit.MILLISECONDS)
+ .until(() -> taskManager1.await(taskId).getStatus() == TaskManager.Status.COMPLETED);
+
+ TaskExecutionDetails detailsFromTaskManager1 = taskManager1.getExecutionDetails(taskId);
+ TaskExecutionDetails detailsFromTaskManager2 = taskManager2.getExecutionDetails(taskId);
+ assertThat(detailsFromTaskManager1).isEqualTo(detailsFromTaskManager2);
+ }
}
@Test
@@ -133,40 +137,45 @@ class DistributedTaskManagerTest implements TaskManagerContract {
}
@Test
- void givenTwoTaskManagersAndTwoTasksOnlyOneTaskShouldRunAtTheSameTime() throws InterruptedException {
+ void givenTwoTaskManagersAndTwoTasksOnlyOneTaskShouldRunAtTheSameTime() {
CountDownLatch waitingForFirstTaskLatch = new CountDownLatch(1);
- TaskManager taskManager1 = taskManager();
- TaskManager taskManager2 = taskManager();
+ try (EventSourcingTaskManager taskManager1 = taskManager();
+ EventSourcingTaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME_2)) {
- taskManager1.submit(new MemoryReferenceTask(() -> {
- waitingForFirstTaskLatch.await();
- return Task.Result.COMPLETED;
- }));
- TaskId waitingTaskId = taskManager1.submit(new CompletedTask());
+ taskManager1.submit(new MemoryReferenceTask(() -> {
+ waitingForFirstTaskLatch.await();
+ return Task.Result.COMPLETED;
+ }));
+ TaskId waitingTaskId = taskManager1.submit(new CompletedTask());
- awaitUntilTaskHasStatus(waitingTaskId, TaskManager.Status.WAITING, taskManager2);
- waitingForFirstTaskLatch.countDown();
+ awaitUntilTaskHasStatus(waitingTaskId, TaskManager.Status.WAITING, taskManager2);
+ waitingForFirstTaskLatch.countDown();
- Awaitility.await()
- .atMost(Duration.ONE_SECOND)
- .pollInterval(100L, TimeUnit.MILLISECONDS)
- .until(() -> taskManager1.await(waitingTaskId).getStatus() == TaskManager.Status.COMPLETED);
+ Awaitility.await()
+ .atMost(Duration.ONE_SECOND)
+ .pollInterval(100L, TimeUnit.MILLISECONDS)
+ .until(() -> taskManager1.await(waitingTaskId).getStatus() == TaskManager.Status.COMPLETED);
+ }
}
@Test
- // FIXME it's currently dependent of the implementation of the sequential TaskManager with the exclusive RabbitMQ consumer
- // once we store the node where the event have been created/started/completed we should rewrite it with this information.
void givenTwoTaskManagerATaskSubmittedOnOneCouldBeRunOnTheOther() throws InterruptedException {
- TaskManager taskManager1 = taskManager();
- Thread.sleep(100); // FIXME used to ensure that taskManager1 is the worker consuming from rabbit
- TaskManager taskManager2 = taskManager();
-
- TaskId taskId = taskManager2.submit(new CompletedTask());
-
- Awaitility.await()
- .atMost(Duration.ONE_SECOND)
- .pollInterval(100L, TimeUnit.MILLISECONDS)
- .until(() -> taskManager1.await(taskId).getStatus() == TaskManager.Status.COMPLETED);
+ try (EventSourcingTaskManager taskManager1 = taskManager()) {
+ Thread.sleep(100);
+ try (EventSourcingTaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME_2)) {
+
+ TaskId taskId = taskManager2.submit(new CompletedTask());
+
+ Awaitility.await()
+ .atMost(Duration.ONE_SECOND)
+ .pollInterval(100L, TimeUnit.MILLISECONDS)
+ .until(() -> taskManager1.await(taskId).getStatus() == TaskManager.Status.COMPLETED);
+
+ TaskExecutionDetails executionDetails = taskManager2.getExecutionDetails(taskId);
+ assertThat(executionDetails.getSubmittedNode()).isEqualTo(HOSTNAME_2);
+ assertThat(executionDetails.getRanNode()).contains(HOSTNAME);
+ }
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/06: JAMES-2813 Implement TaskManagerContract for
DistributedTaskManager
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 5ed834b661015abb71da11d30fee679efd395f77
Author: Rémi Kowalski <rk...@linagora.com>
AuthorDate: Thu Aug 22 12:04:54 2019 +0200
JAMES-2813 Implement TaskManagerContract for DistributedTaskManager
---
.../apache/james/DistributedTaskManagerModule.java | 12 +---
.../distributed/RabbitMQWorkQueueSupplier.scala | 38 +++++++++++++
.../distributed/DistributedTaskManagerTest.java | 66 +++++++++++++++-------
3 files changed, 86 insertions(+), 30 deletions(-)
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java
index f5cb955..4d2b2ab 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java
@@ -21,25 +21,17 @@
package org.apache.james;
import org.apache.james.modules.server.HostnameModule;
-import org.apache.james.task.MemoryWorkQueue;
-import org.apache.james.task.SerialTaskManagerWorker;
import org.apache.james.task.TaskManager;
-import org.apache.james.task.TaskManagerWorker;
import org.apache.james.task.eventsourcing.EventSourcingTaskManager;
import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection;
import org.apache.james.task.eventsourcing.WorkQueueSupplier;
-import org.apache.james.task.eventsourcing.WorkerStatusListener;
import org.apache.james.task.eventsourcing.cassandra.CassandraTaskExecutionDetailsProjection;
+import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueSupplier;
import com.google.inject.AbstractModule;
import com.google.inject.Scopes;
public class DistributedTaskManagerModule extends AbstractModule {
- public static final WorkQueueSupplier workQueueSupplier = eventSourcingSystem -> {
- WorkerStatusListener listener = new WorkerStatusListener(eventSourcingSystem);
- TaskManagerWorker worker = new SerialTaskManagerWorker(listener);
- return new MemoryWorkQueue(worker);
- };
@Override
protected void configure() {
@@ -49,6 +41,6 @@ public class DistributedTaskManagerModule extends AbstractModule {
bind(WorkQueueSupplier.class).in(Scopes.SINGLETON);
bind(TaskExecutionDetailsProjection.class).to(CassandraTaskExecutionDetailsProjection.class);
bind(TaskManager.class).to(EventSourcingTaskManager.class);
- bind(WorkQueueSupplier.class).toInstance(workQueueSupplier);
+ bind(WorkQueueSupplier.class).to(RabbitMQWorkQueueSupplier.class);
}
}
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueSupplier.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueSupplier.scala
new file mode 100644
index 0000000..fc5d7e2
--- /dev/null
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueSupplier.scala
@@ -0,0 +1,38 @@
+/** **************************************************************
+ * 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.james.task.eventsourcing.distributed
+
+import javax.inject.Inject
+import org.apache.james.backend.rabbitmq.SimpleConnectionPool
+import org.apache.james.eventsourcing.EventSourcingSystem
+import org.apache.james.server.task.json.JsonTaskSerializer
+import org.apache.james.task.eventsourcing.{WorkQueueSupplier, WorkerStatusListener}
+import org.apache.james.task.{SerialTaskManagerWorker, WorkQueue}
+
+@Inject
+class RabbitMQWorkQueueSupplier(private val rabbitMQConnectionPool: SimpleConnectionPool,
+ private val jsonTaskSerializer: JsonTaskSerializer) extends WorkQueueSupplier {
+ override def apply(eventSourcingSystem: EventSourcingSystem): WorkQueue = {
+ val listener = WorkerStatusListener(eventSourcingSystem)
+ val worker = new SerialTaskManagerWorker(listener)
+ val rabbitMQWorkQueue = new RabbitMQWorkQueue(worker, rabbitMQConnectionPool, jsonTaskSerializer)
+ rabbitMQWorkQueue.start()
+ rabbitMQWorkQueue
+ }
+}
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
index 219fd06..7561181 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
@@ -22,6 +22,7 @@ package org.apache.james.task.eventsourcing.distributed;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.james.backend.rabbitmq.RabbitMQExtension;
@@ -35,31 +36,42 @@ import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreEx
import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreModule;
import org.apache.james.eventsourcing.eventstore.cassandra.dto.EventDTOModule;
import org.apache.james.server.task.json.JsonTaskSerializer;
+import org.apache.james.server.task.json.dto.MemoryReferenceTaskStore;
import org.apache.james.server.task.json.dto.TestTaskDTOModules;
import org.apache.james.task.CompletedTask;
-import org.apache.james.task.SerialTaskManagerWorker;
+import org.apache.james.task.CountDownLatchExtension;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskId;
import org.apache.james.task.TaskManager;
-import org.apache.james.task.TaskManagerWorker;
+import org.apache.james.task.TaskManagerContract;
import org.apache.james.task.eventsourcing.EventSourcingTaskManager;
import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection;
import org.apache.james.task.eventsourcing.WorkQueueSupplier;
-import org.apache.james.task.eventsourcing.WorkerStatusListener;
import org.apache.james.task.eventsourcing.cassandra.CassandraTaskExecutionDetailsProjection;
import org.apache.james.task.eventsourcing.cassandra.CassandraTaskExecutionDetailsProjectionDAO;
import org.apache.james.task.eventsourcing.cassandra.CassandraTaskExecutionDetailsProjectionModule;
import org.awaitility.Awaitility;
import org.awaitility.Duration;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import com.github.steveash.guavate.Guavate;
-class DistributedTaskManagerTest {
- private static final JsonTaskSerializer TASK_SERIALIZER = new JsonTaskSerializer(TestTaskDTOModules.COMPLETED_TASK_MODULE);
+@ExtendWith(CountDownLatchExtension.class)
+class DistributedTaskManagerTest implements TaskManagerContract {
+ private static final JsonTaskSerializer TASK_SERIALIZER = new JsonTaskSerializer(
+ TestTaskDTOModules.COMPLETED_TASK_MODULE,
+ TestTaskDTOModules.FAILED_TASK_MODULE,
+ TestTaskDTOModules.THROWING_TASK_MODULE,
+ TestTaskDTOModules.MEMORY_REFERENCE_TASK_MODULE.apply(new MemoryReferenceTaskStore()));
+
+ private static final Hostname HOSTNAME = new Hostname("foo");
+ private static final Hostname HOSTNAME_2 = new Hostname("bar");
private static final Set<EventDTOModule> MODULES = TasksSerializationModule.MODULES.apply(TASK_SERIALIZER).stream().collect(Guavate.toImmutableSet());
static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(
@@ -75,22 +87,26 @@ class DistributedTaskManagerTest {
@RegisterExtension
static CassandraEventStoreExtension eventStoreExtension = new CassandraEventStoreExtension(cassandraCluster, MODULES);
+ private final CassandraCluster cassandra = cassandraCluster.getCassandraCluster();
+ private final CassandraTaskExecutionDetailsProjectionDAO cassandraTaskExecutionDetailsProjectionDAO = new CassandraTaskExecutionDetailsProjectionDAO(cassandra.getConf(), cassandra.getTypesProvider());
+ private final TaskExecutionDetailsProjection executionDetailsProjection = new CassandraTaskExecutionDetailsProjection(cassandraTaskExecutionDetailsProjectionDAO);
+
+ WorkQueueSupplier workQueueSupplier = new RabbitMQWorkQueueSupplier(rabbitMQExtension.getRabbitConnectionPool(), TASK_SERIALIZER);
+ private EventStore eventStore;
+
+ @BeforeEach
+ void setUp(EventStore eventStore) {
+ this.eventStore = eventStore;
+ }
+
+ public TaskManager taskManager() {
+ return new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME);
+ }
+
@Test
- void givenOneEventStoreTwoEventTaskManagersShareTheSameEvents(EventStore eventStore) {
- CassandraCluster cassandra = cassandraCluster.getCassandraCluster();
- CassandraTaskExecutionDetailsProjectionDAO cassandraTaskExecutionDetailsProjectionDAO = new CassandraTaskExecutionDetailsProjectionDAO(cassandra.getConf(), cassandra.getTypesProvider());
- TaskExecutionDetailsProjection executionDetailsProjection = new CassandraTaskExecutionDetailsProjection(cassandraTaskExecutionDetailsProjectionDAO);
-
- WorkQueueSupplier workQueueSupplier = eventSourcingSystem -> {
- WorkerStatusListener listener = new WorkerStatusListener(eventSourcingSystem);
- TaskManagerWorker worker = new SerialTaskManagerWorker(listener);
- RabbitMQWorkQueue rabbitMQWorkQueue = new RabbitMQWorkQueue(worker, rabbitMQExtension.getRabbitConnectionPool(), TASK_SERIALIZER);
- rabbitMQWorkQueue.start();
- return rabbitMQWorkQueue;
-
- };
- TaskManager taskManager1 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, new Hostname("foo"));
- TaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, new Hostname("bar"));
+ void givenOneEventStoreTwoEventTaskManagersShareTheSameEvents() {
+ TaskManager taskManager1 = taskManager();
+ TaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME_2);
TaskId taskId = taskManager1.submit(new CompletedTask());
Awaitility.await()
@@ -102,4 +118,14 @@ class DistributedTaskManagerTest {
TaskExecutionDetails detailsFromTaskManager2 = taskManager2.getExecutionDetails(taskId);
assertThat(detailsFromTaskManager1).isEqualTo(detailsFromTaskManager2);
}
+
+ @Test
+ @Disabled("Cancelling is not supported yet")
+ public void aWaitingTaskShouldBeCancelled(CountDownLatch countDownLatch) {
+ }
+
+ @Test
+ @Disabled("Cancelling is not supported yet")
+ public void getStatusShouldBeCancelledWhenCancelled(CountDownLatch countDownLatch) {
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/06: JAMES-2813 Use RabbitMQWorkqueue in the
distributed TaskManager test
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 292e6dca2b0e2fe3f6745d01d2727c0d773199e6
Author: Rémi Kowalski <rk...@linagora.com>
AuthorDate: Thu Aug 22 11:01:19 2019 +0200
JAMES-2813 Use RabbitMQWorkqueue in the distributed TaskManager test
---
.../eventsourcing/distributed/DistributedTaskManagerTest.java | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
index d84243c..219fd06 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
@@ -24,6 +24,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import org.apache.james.backend.rabbitmq.RabbitMQExtension;
import org.apache.james.backends.cassandra.CassandraCluster;
import org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.backends.cassandra.components.CassandraModule;
@@ -36,7 +37,6 @@ import org.apache.james.eventsourcing.eventstore.cassandra.dto.EventDTOModule;
import org.apache.james.server.task.json.JsonTaskSerializer;
import org.apache.james.server.task.json.dto.TestTaskDTOModules;
import org.apache.james.task.CompletedTask;
-import org.apache.james.task.MemoryWorkQueue;
import org.apache.james.task.SerialTaskManagerWorker;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskId;
@@ -70,6 +70,9 @@ class DistributedTaskManagerTest {
CassandraTaskExecutionDetailsProjectionModule.MODULE()));
@RegisterExtension
+ static RabbitMQExtension rabbitMQExtension = RabbitMQExtension.singletonRabbitMQ();
+
+ @RegisterExtension
static CassandraEventStoreExtension eventStoreExtension = new CassandraEventStoreExtension(cassandraCluster, MODULES);
@Test
@@ -81,7 +84,10 @@ class DistributedTaskManagerTest {
WorkQueueSupplier workQueueSupplier = eventSourcingSystem -> {
WorkerStatusListener listener = new WorkerStatusListener(eventSourcingSystem);
TaskManagerWorker worker = new SerialTaskManagerWorker(listener);
- return new MemoryWorkQueue(worker);
+ RabbitMQWorkQueue rabbitMQWorkQueue = new RabbitMQWorkQueue(worker, rabbitMQExtension.getRabbitConnectionPool(), TASK_SERIALIZER);
+ rabbitMQWorkQueue.start();
+ return rabbitMQWorkQueue;
+
};
TaskManager taskManager1 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, new Hostname("foo"));
TaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, new Hostname("bar"));
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 06/06: JAMES-2813 Fix distributed task manager
runtime dependency
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit fda3446d719b2129ed9b4c92ec989d56df3b2e02
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Mon Sep 2 14:33:34 2019 +0200
JAMES-2813 Fix distributed task manager runtime dependency
---
server/task-distributed/pom.xml | 18 ++++++++++++++----
1 file changed, 14 insertions(+), 4 deletions(-)
diff --git a/server/task-distributed/pom.xml b/server/task-distributed/pom.xml
index 89b4edf..56e58a8 100644
--- a/server/task-distributed/pom.xml
+++ b/server/task-distributed/pom.xml
@@ -39,17 +39,17 @@
</dependency>
<dependency>
<groupId>${james.groupId}</groupId>
- <artifactId>apache-james-backends-rabbitmq</artifactId>
+ <artifactId>apache-james-backends-cassandra</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>${james.groupId}</groupId>
<artifactId>apache-james-backends-rabbitmq</artifactId>
- <type>test-jar</type>
- <scope>test</scope>
</dependency>
<dependency>
<groupId>${james.groupId}</groupId>
- <artifactId>apache-james-backends-cassandra</artifactId>
+ <artifactId>apache-james-backends-rabbitmq</artifactId>
<type>test-jar</type>
<scope>test</scope>
</dependency>
@@ -99,6 +99,11 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-testing</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
@@ -113,6 +118,11 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org