You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by GitBox <gi...@apache.org> on 2022/06/23 11:01:15 UTC
[GitHub] [james-project] vttranlina opened a new pull request, #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
vttranlina opened a new pull request, #1062:
URL: https://github.com/apache/james-project/pull/1062
Jira: https://issues.apache.org/jira/projects/JAMES/issues/JAMES-3784
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] chibenwa commented on a diff in pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
chibenwa commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r905962989
##########
server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java:
##########
@@ -78,35 +82,51 @@ void taskShouldThrowOnDeserializationUrlDecodingError(String serialized) {
@Test
void additionalInformationShouldBeSerializable() throws Exception {
ReprocessingAllMailsTask.AdditionalInformation details = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME),
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
ReprocessingAllMailsTask.AdditionalInformation detailsWithoutProcessor = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME),
+ new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME, Limit.unlimited()),
+ REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
+
+ ReprocessingAllMailsTask.AdditionalInformation detailWithLimit = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.limit(10)),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
JsonSerializationVerifier.dtoModule(ReprocessingAllMailsTaskAdditionalInformationDTO.module())
.testCase(details, SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITH_TARGET_PROCESSOR)
.testCase(detailsWithoutProcessor, SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITHOUT_TARGET_PROCESSOR)
+ .testCase(detailWithLimit, "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\",\"consume\":true, \"limit\": 10}")
.verify();
}
@Test
void shouldDeserializePreviousTaskFormat() throws Exception {
- ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME));
+ ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()));
JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingAllMailsTaskDTO.module(REPROCESSING_SERVICE));
assertThat(testee.deserialize(OLD_SERIALIZED_TASK))
.isEqualToComparingFieldByFieldRecursively(taskWithTargetProcessor);
+
+ String serializedTaskWithLimit = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\", \"limit\":10}";
Review Comment:
Do we stiff support JSONs without limit defined?
##########
server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverTaskTest.java:
##########
@@ -64,6 +66,13 @@ void redeliverAllTaskShouldMatchJsonSerializationContract() throws Exception {
.bean(TASK_ALL)
.json(SERIALIZED_ALL)
.verify();
+
+ EventDeadLettersRedeliverAllTask taskAllWithLimit = new EventDeadLettersRedeliverAllTask(SERVICE, new RunningOptions(Limit.limit(10)));
+
+ JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverAllTaskDTO.module(SERVICE))
+ .bean(taskAllWithLimit)
+ .json("{\"type\":\"event-dead-letters-redeliver-all\", \"runningOptions\":{\"limit\": 10}}")
Review Comment:
Can we still deserialize a JSON without runningOptions ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] chibenwa commented on a diff in pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
chibenwa commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r905213978
##########
server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java:
##########
@@ -0,0 +1,45 @@
+/****************************************************************
+ * 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.modules;
+
+import static org.apache.james.mailrepository.api.MailRepositoryEmptyHealthCheck.ErrorMailRepositoryEmptyHealthCheck;
+
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.mailrepository.api.MailRepositoryStore;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.multibindings.Multibinder;
+
+public class ErrorMailRepositoryEmptyHealthCheckModule extends AbstractModule {
Review Comment:
EmptyErrorMailRepositoryHealthCheckModule
##########
server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/MailRepositoryEmptyHealthCheck.java:
##########
@@ -0,0 +1,71 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public abstract class MailRepositoryEmptyHealthCheck implements HealthCheck {
+
+ protected ComponentName componentName;
+ private final MailRepositoryStore repositoryStore;
+ private final MailRepositoryPath path;
+
+ public MailRepositoryEmptyHealthCheck(MailRepositoryPath path, MailRepositoryStore repositoryStore) {
+ this.repositoryStore = repositoryStore;
+ this.path = path;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return componentName;
+ }
+
+ @Override
+ public Publisher<Result> check() {
+ // TODO: use .sizeReactive of https://github.com/apache/james-project/pull/1049/
+ return Flux.fromStream(Throwing.supplier(() -> repositoryStore.getByPath(path)))
+ .any(Throwing.predicate(repository -> repository.size() > 0))
+ .filter(FunctionalUtils.identityPredicate())
+ .map(hasSize -> Result.degraded(componentName, "MailRepository is not empty"))
+ .switchIfEmpty(Mono.just(Result.healthy(componentName)));
+ }
+
+ public static class ErrorMailRepositoryEmptyHealthCheck extends MailRepositoryEmptyHealthCheck {
+ private static final ComponentName COMPONENT_NAME = new ComponentName("ErrorMailRepositoryEmpty");
+
+ @Override
+ public ComponentName componentName() {
+ return COMPONENT_NAME;
+ }
+
+ public ErrorMailRepositoryEmptyHealthCheck(MailRepositoryStore repositoryStore) {
+ super(MailRepositoryPath.from("var/mail/error"), repositoryStore);
Review Comment:
We should not hard code this but rather retrieve what is configured in JamesMailSpooler.
##########
server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/routes/EventDeadLettersRoutes.java:
##########
@@ -144,4 +150,11 @@ private EventDeadLetters.InsertionId parseInsertionId(Request request) {
.haltError();
}
}
+
+ private RunningOptions parseRunningOptions(Request request) {
+ return ParametersExtractor.extractPositiveInteger(request, "rateLimit")
+ .map(rateLimit -> new RunningOptions(rateLimit, ParametersExtractor.extractDuration(request, "rateLimitWindow")
+ .orElse(Duration.ofSeconds(1))))
+ .orElse(RunningOptions.DEFAULT);
+ }
Review Comment:
No we are not defining a rate limit (implemented with a throttler) but a **limit** (implemented by a limit method on java streams and take on flux).
-> `rateLimit` -> `limit`
-> No duration
##########
server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/MailRepositoryEmptyHealthCheck.java:
##########
@@ -0,0 +1,71 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public abstract class MailRepositoryEmptyHealthCheck implements HealthCheck {
+
+ protected ComponentName componentName;
+ private final MailRepositoryStore repositoryStore;
+ private final MailRepositoryPath path;
+
+ public MailRepositoryEmptyHealthCheck(MailRepositoryPath path, MailRepositoryStore repositoryStore) {
+ this.repositoryStore = repositoryStore;
+ this.path = path;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return componentName;
+ }
+
+ @Override
+ public Publisher<Result> check() {
+ // TODO: use .sizeReactive of https://github.com/apache/james-project/pull/1049/
+ return Flux.fromStream(Throwing.supplier(() -> repositoryStore.getByPath(path)))
+ .any(Throwing.predicate(repository -> repository.size() > 0))
+ .filter(FunctionalUtils.identityPredicate())
+ .map(hasSize -> Result.degraded(componentName, "MailRepository is not empty"))
+ .switchIfEmpty(Mono.just(Result.healthy(componentName)));
+ }
+
+ public static class ErrorMailRepositoryEmptyHealthCheck extends MailRepositoryEmptyHealthCheck {
Review Comment:
IMO inheritence here is evil.
What problem does it tries to solve? Alternative approaches?
##########
server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java:
##########
@@ -120,19 +144,32 @@ public ReprocessingService(MailQueueFactory<?> mailQueueFactory,
this.mailRepositoryStoreService = mailRepositoryStoreService;
}
- public void reprocessAll(MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
- try (Reprocessor reprocessor = new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration)) {
- mailRepositoryStoreService
- .getRepositories(path)
- .forEach(Throwing.consumer((MailRepository repository) ->
- Iterators.toStream(repository.list())
- .peek(keyListener)
- .forEach(Throwing.consumer(key ->
- Optional.ofNullable(repository.retrieve(key))
- .ifPresent(mail -> reprocessor.reprocess(repository, mail, key))
- ))
- ));
- }
+ public Mono<Task.Result> reprocessAll(MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) {
+ return Mono.using(() -> new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration),
+ reprocessor -> reprocessAll(reprocessor, path, configuration, keyListener),
+ Reprocessor::close);
+ }
+
+ private Mono<Task.Result> reprocessAll(Reprocessor reprocessor, MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) {
+ return Flux.fromStream(Throwing.supplier(() -> mailRepositoryStoreService.getRepositories(path)))
+ .flatMap(Throwing.function((MailRepository repository) -> Iterators.toFlux(repository.list())
+ .doOnNext(keyListener)
+ .map(mailKey -> Pair.of(repository, mailKey))))
+ .transform(ReactorUtils.<Pair<MailRepository, MailKey>, Task.Result>throttle()
+ .elements(configuration.getRateLimit())
+ .per(configuration.getRateLimitUnit())
+ .forOperation(pair -> reprocess(pair.getRight(), pair.getLeft(), reprocessor)))
+ .reduce(Task.Result.COMPLETED, Task::combine);
Review Comment:
Idem a .take does the job, not throttling.
##########
server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/utils/ParametersExtractor.java:
##########
@@ -51,6 +54,12 @@ public static Optional<Integer> extractPositiveInteger(Request request, String p
return extractPositiveNumber(request, parameterName, Integer::valueOf);
}
+ public static Optional<Duration> extractDuration(Request request, String parameterName) {
+ return Optional.ofNullable(request.queryParams(parameterName))
+ .filter(s -> !Strings.isNullOrEmpty(s))
Review Comment:
Can't be null so `.filter(s -> !s.isEmpty())` is enough?
##########
server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/MailRepositoryEmptyHealthCheck.java:
##########
@@ -0,0 +1,71 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public abstract class MailRepositoryEmptyHealthCheck implements HealthCheck {
+
+ protected ComponentName componentName;
+ private final MailRepositoryStore repositoryStore;
+ private final MailRepositoryPath path;
+
+ public MailRepositoryEmptyHealthCheck(MailRepositoryPath path, MailRepositoryStore repositoryStore) {
+ this.repositoryStore = repositoryStore;
+ this.path = path;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return componentName;
+ }
+
+ @Override
+ public Publisher<Result> check() {
+ // TODO: use .sizeReactive of https://github.com/apache/james-project/pull/1049/
+ return Flux.fromStream(Throwing.supplier(() -> repositoryStore.getByPath(path)))
+ .any(Throwing.predicate(repository -> repository.size() > 0))
+ .filter(FunctionalUtils.identityPredicate())
+ .map(hasSize -> Result.degraded(componentName, "MailRepository is not empty"))
+ .switchIfEmpty(Mono.just(Result.healthy(componentName)));
+ }
+
+ public static class ErrorMailRepositoryEmptyHealthCheck extends MailRepositoryEmptyHealthCheck {
+ private static final ComponentName COMPONENT_NAME = new ComponentName("ErrorMailRepositoryEmpty");
Review Comment:
EmptyErrorMailRepository ?
##########
server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/MailRepositoryEmptyHealthCheck.java:
##########
@@ -0,0 +1,71 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public abstract class MailRepositoryEmptyHealthCheck implements HealthCheck {
Review Comment:
EmptyMailRepositoryHealthCheck
##########
server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java:
##########
@@ -120,19 +144,32 @@ public ReprocessingService(MailQueueFactory<?> mailQueueFactory,
this.mailRepositoryStoreService = mailRepositoryStoreService;
}
- public void reprocessAll(MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
- try (Reprocessor reprocessor = new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration)) {
- mailRepositoryStoreService
- .getRepositories(path)
- .forEach(Throwing.consumer((MailRepository repository) ->
- Iterators.toStream(repository.list())
- .peek(keyListener)
- .forEach(Throwing.consumer(key ->
- Optional.ofNullable(repository.retrieve(key))
- .ifPresent(mail -> reprocessor.reprocess(repository, mail, key))
- ))
- ));
- }
+ public Mono<Task.Result> reprocessAll(MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) {
+ return Mono.using(() -> new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration),
+ reprocessor -> reprocessAll(reprocessor, path, configuration, keyListener),
+ Reprocessor::close);
+ }
+
+ private Mono<Task.Result> reprocessAll(Reprocessor reprocessor, MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) {
+ return Flux.fromStream(Throwing.supplier(() -> mailRepositoryStoreService.getRepositories(path)))
Review Comment:
Any chances this does blocking reads?
##########
server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java:
##########
@@ -75,6 +91,14 @@ public Optional<String> getTargetProcessor() {
public boolean isConsume() {
return consume;
}
+
+ public int getRateLimit() {
+ return rateLimit;
+ }
+
+ public Duration getRateLimitUnit() {
+ return rateLimitUnit;
+ }
Review Comment:
```suggestion
public Limit getLimit() {
return limit;
}
```
(We already should have a nice POJO objects for limit
##########
server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java:
##########
@@ -109,13 +107,9 @@ private void notifyProgress(MailKey key) {
@Override
public Result run() {
- try {
- reprocessingService.reprocessAll(repositoryPath, configuration, this::notifyProgress);
- return Result.COMPLETED;
- } catch (MessagingException | MailRepositoryStore.MailRepositoryStoreException e) {
- LOGGER.error("Encountered error while reprocessing repository", e);
- return Result.PARTIAL;
- }
+ return reprocessingService.reprocessAll(repositoryPath, configuration, this::notifyProgress)
+ .subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER)
+ .block();
Review Comment:
```suggestion
return reprocessingService.reprocessAll(repositoryPath, configuration, this::notifyProgress)
.block();
```
Please no subscribeOn before .block.
We have a thread at hand that sits doing nothing waiting for something to completes, we should use it to cary other the execution rather than pulling yet another thread.
##########
server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverService.java:
##########
@@ -52,9 +71,12 @@ public EventDeadLettersRedeliverService(Set<EventBus> eventBuses, EventDeadLette
this.deadLetters = deadLetters;
}
- Flux<Task.Result> redeliverEvents(EventRetriever eventRetriever) {
+ Flux<Task.Result> redeliverEvents(EventRetriever eventRetriever, RunningOptions runningOptions) {
return eventRetriever.retrieveEvents(deadLetters)
- .flatMap(entry -> redeliverGroupEvents(entry.getT1(), entry.getT2(), entry.getT3()), DEFAULT_CONCURRENCY);
+ .transform(ReactorUtils.<Tuple3<Group, Event, EventDeadLetters.InsertionId>, Task.Result>throttle()
+ .elements(runningOptions.getRateLimit())
+ .per(runningOptions.getRateLimitWindow())
+ .forOperation(tuple3 -> redeliverGroupEvents(tuple3.getT1(), tuple3.getT2(), tuple3.getT3())));
Review Comment:
```
return eventRetriever.retrieveEvents(deadLetters)
.take(runningOptions.getLimit())
.flatMap(entry -> redeliverGroupEvents(entry.getT1(), entry.getT2(), entry.getT3()), DEFAULT_CONCURRENCY);
```
Of course if the limit is defined.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] vttranlina commented on a diff in pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
vttranlina commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r905964246
##########
server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java:
##########
@@ -78,35 +82,51 @@ void taskShouldThrowOnDeserializationUrlDecodingError(String serialized) {
@Test
void additionalInformationShouldBeSerializable() throws Exception {
ReprocessingAllMailsTask.AdditionalInformation details = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME),
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
ReprocessingAllMailsTask.AdditionalInformation detailsWithoutProcessor = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME),
+ new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME, Limit.unlimited()),
+ REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
+
+ ReprocessingAllMailsTask.AdditionalInformation detailWithLimit = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.limit(10)),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
JsonSerializationVerifier.dtoModule(ReprocessingAllMailsTaskAdditionalInformationDTO.module())
.testCase(details, SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITH_TARGET_PROCESSOR)
.testCase(detailsWithoutProcessor, SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITHOUT_TARGET_PROCESSOR)
+ .testCase(detailWithLimit, "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\",\"consume\":true, \"limit\": 10}")
.verify();
}
@Test
void shouldDeserializePreviousTaskFormat() throws Exception {
- ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME));
+ ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()));
JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingAllMailsTaskDTO.module(REPROCESSING_SERVICE));
assertThat(testee.deserialize(OLD_SERIALIZED_TASK))
.isEqualToComparingFieldByFieldRecursively(taskWithTargetProcessor);
+
+ String serializedTaskWithLimit = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\", \"limit\":10}";
Review Comment:
yes, it is the default,
It was checked at line 107
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] vttranlina commented on pull request #1062: JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
vttranlina commented on PR #1062:
URL: https://github.com/apache/james-project/pull/1062#issuecomment-1169503694
Squash fixup & Complete TODO
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] chibenwa commented on a diff in pull request #1062: JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
chibenwa commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r908260742
##########
server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheck.java:
##########
@@ -0,0 +1,56 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class EmptyErrorMailRepositoryHealthCheck implements HealthCheck {
+ public static final ComponentName COMPONENT_NAME = new ComponentName("EmptyErrorMailRepository");
+ private final MailRepositoryStore repositoryStore;
+ private final MailRepositoryPath errorRepositoryPath;
+
+ public EmptyErrorMailRepositoryHealthCheck(MailRepositoryPath errorRepositoryPath, MailRepositoryStore repositoryStore) {
+ this.repositoryStore = repositoryStore;
+ this.errorRepositoryPath = errorRepositoryPath;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return COMPONENT_NAME;
+ }
+
+ @Override
+ public Mono<Result> check() {
+ // TODO: use .sizeReactive of https://github.com/apache/james-project/pull/1049/
Review Comment:
Mmh TODO: getting rid of the TODO
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] vttranlina commented on a diff in pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
vttranlina commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r905965754
##########
server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverTaskTest.java:
##########
@@ -64,6 +66,13 @@ void redeliverAllTaskShouldMatchJsonSerializationContract() throws Exception {
.bean(TASK_ALL)
.json(SERIALIZED_ALL)
.verify();
+
+ EventDeadLettersRedeliverAllTask taskAllWithLimit = new EventDeadLettersRedeliverAllTask(SERVICE, new RunningOptions(Limit.limit(10)));
+
+ JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverAllTaskDTO.module(SERVICE))
+ .bean(taskAllWithLimit)
+ .json("{\"type\":\"event-dead-letters-redeliver-all\", \"runningOptions\":{\"limit\": 10}}")
Review Comment:
It will handle case old tasks already in the database, without runningOptions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] vttranlina commented on a diff in pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
vttranlina commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r905684557
##########
server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/MailRepositoryEmptyHealthCheck.java:
##########
@@ -0,0 +1,71 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public abstract class MailRepositoryEmptyHealthCheck implements HealthCheck {
+
+ protected ComponentName componentName;
+ private final MailRepositoryStore repositoryStore;
+ private final MailRepositoryPath path;
+
+ public MailRepositoryEmptyHealthCheck(MailRepositoryPath path, MailRepositoryStore repositoryStore) {
+ this.repositoryStore = repositoryStore;
+ this.path = path;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return componentName;
+ }
+
+ @Override
+ public Publisher<Result> check() {
+ // TODO: use .sizeReactive of https://github.com/apache/james-project/pull/1049/
+ return Flux.fromStream(Throwing.supplier(() -> repositoryStore.getByPath(path)))
+ .any(Throwing.predicate(repository -> repository.size() > 0))
+ .filter(FunctionalUtils.identityPredicate())
+ .map(hasSize -> Result.degraded(componentName, "MailRepository is not empty"))
+ .switchIfEmpty(Mono.just(Result.healthy(componentName)));
+ }
+
+ public static class ErrorMailRepositoryEmptyHealthCheck extends MailRepositoryEmptyHealthCheck {
Review Comment:
IMO in feature, we can easy to create a health check for any "repository path", not only `var/error`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] chibenwa commented on pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
chibenwa commented on PR #1062:
URL: https://github.com/apache/james-project/pull/1062#issuecomment-1166196811
```
11:08:53,643 [ERROR] Failed to execute goal org.apache.maven.plugins:maven-checkstyle-plugin:3.1.0:check (check-style) on project james-server-webadmin-mailrepository: You have 2 Checkstyle violations. -> [Help 1]
org.apache.maven.lifecycle.LifecycleExecutionException: Failed to execute goal org.apache.maven.plugins:maven-checkstyle-plugin:3.1.0:check (check-style) on project james-server-webadmin-mailrepository: You have 2 Checkstyle violations.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] Arsnael merged pull request #1062: JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
Arsnael merged PR #1062:
URL: https://github.com/apache/james-project/pull/1062
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] chibenwa commented on a diff in pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
chibenwa commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r905968504
##########
server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverTaskTest.java:
##########
@@ -64,6 +66,13 @@ void redeliverAllTaskShouldMatchJsonSerializationContract() throws Exception {
.bean(TASK_ALL)
.json(SERIALIZED_ALL)
.verify();
+
+ EventDeadLettersRedeliverAllTask taskAllWithLimit = new EventDeadLettersRedeliverAllTask(SERVICE, new RunningOptions(Limit.limit(10)));
+
+ JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverAllTaskDTO.module(SERVICE))
+ .bean(taskAllWithLimit)
+ .json("{\"type\":\"event-dead-letters-redeliver-all\", \"runningOptions\":{\"limit\": 10}}")
Review Comment:
I mean, do we have a test for this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] chibenwa commented on a diff in pull request #1062: [WIP] JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
chibenwa commented on code in PR #1062:
URL: https://github.com/apache/james-project/pull/1062#discussion_r905714841
##########
server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/MailRepositoryEmptyHealthCheck.java:
##########
@@ -0,0 +1,71 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public abstract class MailRepositoryEmptyHealthCheck implements HealthCheck {
+
+ protected ComponentName componentName;
+ private final MailRepositoryStore repositoryStore;
+ private final MailRepositoryPath path;
+
+ public MailRepositoryEmptyHealthCheck(MailRepositoryPath path, MailRepositoryStore repositoryStore) {
+ this.repositoryStore = repositoryStore;
+ this.path = path;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return componentName;
+ }
+
+ @Override
+ public Publisher<Result> check() {
+ // TODO: use .sizeReactive of https://github.com/apache/james-project/pull/1049/
+ return Flux.fromStream(Throwing.supplier(() -> repositoryStore.getByPath(path)))
+ .any(Throwing.predicate(repository -> repository.size() > 0))
+ .filter(FunctionalUtils.identityPredicate())
+ .map(hasSize -> Result.degraded(componentName, "MailRepository is not empty"))
+ .switchIfEmpty(Mono.just(Result.healthy(componentName)));
+ }
+
+ public static class ErrorMailRepositoryEmptyHealthCheck extends MailRepositoryEmptyHealthCheck {
Review Comment:
Favor composition other inheritance then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[GitHub] [james-project] Arsnael commented on pull request #1062: JAMES-3784 Ease mail repository / event dead letter operation
Posted by GitBox <gi...@apache.org>.
Arsnael commented on PR #1062:
URL: https://github.com/apache/james-project/pull/1062#issuecomment-1196196964
You can rebase it again (sorry) as I just merged https://github.com/apache/james-project/pull/1049
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org