Skip to content

Commit

Permalink
[SDCISA-16293] Revert everything back to develop, because test error …
Browse files Browse the repository at this point in the history 
…messages as totally useless.
  • Loading branch information
@hiddenalpha
hiddenalpha committed Jun 27, 2024
1 parent 81ab362 commit fe125e2
Show file tree
Hide file tree
Showing 13 changed files with 41 additions and 287 deletions.
45 changes: 14 additions & 31 deletions src/main/java/org/swisspush/redisques/RedisQues.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import org.swisspush.redisques.action.QueueAction;
import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
import org.swisspush.redisques.handler.RedisquesHttpRequestHandler;
import org.swisspush.redisques.performance.BurstSquasher;
import org.swisspush.redisques.performance.UpperBoundParallel;
import org.swisspush.redisques.scheduling.PeriodicSkipScheduler;
import org.swisspush.redisques.util.*;
Expand All @@ -40,7 +39,6 @@
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

Expand Down Expand Up @@ -230,13 +228,14 @@ private enum QueueState {
private final Semaphore checkQueueRequestsQuota;
private final Semaphore queueStatsRequestQuota;
private final Semaphore getQueuesItemsCountRedisRequestQuota;
private static final AtomicReference<BurstSquasher<Object[]>> checkExistsFailLogger = new AtomicReference<>();
private static final AtomicReference<BurstSquasher<Throwable>> checkQueueOwnerLogger = new AtomicReference<>();

public RedisQues() {
this(null, null, null, newThriftyExceptionFactory(), new Semaphore(Integer.MAX_VALUE),
new Semaphore(Integer.MAX_VALUE), new Semaphore(Integer.MAX_VALUE), new Semaphore(Integer.MAX_VALUE));
this.exceptionFactory = newThriftyExceptionFactory();
log.warn("Fallback to legacy behavior and allow up to {} simultaneous requests to redis", Integer.MAX_VALUE);
this.redisMonitoringReqQuota = new Semaphore(Integer.MAX_VALUE);
this.checkQueueRequestsQuota = new Semaphore(Integer.MAX_VALUE);
this.queueStatsRequestQuota = new Semaphore(Integer.MAX_VALUE);
this.getQueuesItemsCountRedisRequestQuota = new Semaphore(Integer.MAX_VALUE);
}

public RedisQues(
Expand Down Expand Up @@ -344,7 +343,7 @@ public void start(Promise<Void> promise) {
redisProvider = new DefaultRedisProvider(vertx, configurationProvider);
}

this.upperBoundParallel = new UpperBoundParallel(vertx, exceptionFactory);
this.upperBoundParallel = new UpperBoundParallel(vertx);

redisProvider.redis().onComplete(event -> {
if(event.succeeded()) {
Expand Down Expand Up @@ -550,11 +549,7 @@ private void registerActiveQueueRegistrationRefresh() {
return iter.hasNext();
}
@Override public boolean onError(Throwable ex, Iterator<Map.Entry<String, QueueState>> iter) {
checkQueueOwnerLogger.compareAndSet(null, new BurstSquasher<>(vertx, (int count, Throwable ex_) -> {
log.warn("TODO error handling", ex_);
}));
checkQueueOwnerLogger.get().logSomewhen(exceptionFactory.newException(ex));
onPeriodicDone.run();
if (log.isWarnEnabled()) log.warn("TODO error handling", exceptionFactory.newException(ex));
return false;
}
@Override public void onDone(Iterator<Map.Entry<String, QueueState>> iter) {
Expand Down Expand Up @@ -1198,12 +1193,8 @@ private Future<Void> checkQueues() {
};
ctx.redisAPI.exists(Collections.singletonList(key), event -> {
if (event.failed() || event.result() == null) {
checkExistsFailLogger.compareAndSet(null, new BurstSquasher<>(vertx, (int count, Object[] detail) -> {
log.error("RedisQues was {} times unable to check existence of some queue like {}",
count, detail[0], (Throwable)detail[1]);
}));
checkExistsFailLogger.get().logSomewhen(new Object[]{
queueName, exceptionFactory.newException("redisAPI.exists(...) failed", event.cause())});
log.error("RedisQues is unable to check existence of queue " + queueName,
exceptionFactory.newException("redisAPI.exists(" + key + ") failed", event.cause()));
onDone.accept(null, null);
return;
}
Expand Down Expand Up @@ -1258,24 +1249,16 @@ private Future<Void> checkQueues() {
return ctx.iter.hasNext();
}
@Override public boolean onError(Throwable ex, Void ctx_) {
// Old code [1] for whatever reason did just continue after an error.
// Nevertheless, I'll now change it to abort (aka return false). Due to
// the absence of comments I have no idea what those half-baked results
// where used for. See also [2] about this decision.
// [1]: https://github.com/swisspost/vertx-redisques/blob/v3.0.33/src/main/java/org/swisspush/redisques/RedisQues.java
// [2]: https://github.com/swisspost/vertx-redisques/issues/192
reduceGcGraphTraversalEffort();
p.fail(exceptionFactory.newException(ex));
return false;
log.warn("TODO error handling", exceptionFactory.newException(ex));
return true; // true, keep going with other queues.
}
@Override public void onDone(Void ctx_) {
reduceGcGraphTraversalEffort();
p.complete(); // Mark this composition step as completed.
}
private void reduceGcGraphTraversalEffort() {
// No longer used, so reduce GC graph traversal effort.
ctx.redisAPI = null;
ctx.counter = null;
ctx.iter = null;
// Mark this composition step as completed.
p.complete();
}
});
return p.future();
Expand Down
8 changes: 2 additions & 6 deletions src/main/java/org/swisspush/redisques/action/GetQueuesCountAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonObject;
import org.slf4j.Logger;
import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
import org.swisspush.redisques.handler.GetQueuesCountHandler;
import org.swisspush.redisques.util.*;

Expand All @@ -16,17 +15,14 @@

public class GetQueuesCountAction extends GetQueuesAction {

private final RedisQuesExceptionFactory exceptionFactory;

public GetQueuesCountAction(
Vertx vertx, RedisProvider redisProvider, RedisQuesExceptionFactory exceptionFactory,
String address, String queuesKey, String queuesPrefix, String consumersPrefix, String locksKey,
Vertx vertx, RedisProvider redisProvider, String address, String queuesKey,
String queuesPrefix, String consumersPrefix, String locksKey,
List<QueueConfiguration> queueConfigurations, QueueStatisticsCollector queueStatisticsCollector,
Logger log
) {
super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
queueConfigurations, queueStatisticsCollector, log);
this.exceptionFactory = exceptionFactory;
}

@Override
Expand Down
2 changes: 0 additions & 2 deletions src/main/java/org/swisspush/redisques/exception/RedisQuesExceptionFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,4 @@ public static RedisQuesExceptionFactory newWastefulExceptionFactory() {
return new WastefulRedisQuesExceptionFactory();
}

ResourceExhaustionException newResourceExhaustionException(String msg, Throwable cause);

}
16 changes: 0 additions & 16 deletions src/main/java/org/swisspush/redisques/exception/ResourceExhaustionException.java
View file

This file was deleted.

8 changes: 0 additions & 8 deletions src/main/java/org/swisspush/redisques/exception/ThriftyRedisQuesExceptionFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,12 +37,4 @@ public ReplyException newReplyException(ReplyFailure failureType, int failureCod
return new NoStackReplyException(failureType, failureCode, msg);
}

@Override
public ResourceExhaustionException newResourceExhaustionException(String msg, Throwable cause) {
if (cause instanceof ResourceExhaustionException) return (ResourceExhaustionException) cause;
return new ResourceExhaustionException(msg, cause) {
@Override public Throwable fillInStackTrace() { return this; }
};
}

}
5 changes: 0 additions & 5 deletions src/main/java/org/swisspush/redisques/exception/WastefulRedisQuesExceptionFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,4 @@ public ReplyException newReplyException(ReplyFailure failureType, int failureCod
return new ReplyException(failureType, failureCode, msg);
}

@Override
public ResourceExhaustionException newResourceExhaustionException(String msg, Throwable cause) {
return new ResourceExhaustionException(msg, cause);
}

}
11 changes: 5 additions & 6 deletions src/main/java/org/swisspush/redisques/handler/GetQueuesItemsCountHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.Message;
import io.vertx.core.eventbus.ReplyException;
import io.vertx.core.eventbus.ReplyFailure;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
Expand All @@ -14,7 +13,6 @@
import io.vertx.redis.client.Response;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
import org.swisspush.redisques.performance.UpperBoundParallel;
import org.swisspush.redisques.util.HandlerUtil;
import org.swisspush.redisques.util.RedisProvider;
Expand All @@ -26,7 +24,8 @@
import java.util.function.BiConsumer;
import java.util.regex.Pattern;

import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
import org.swisspush.redisques.exception.RedisQuesExceptionFactory;

import static java.lang.System.currentTimeMillis;
import static org.swisspush.redisques.util.RedisquesAPI.ERROR;
import static org.swisspush.redisques.util.RedisquesAPI.MONITOR_QUEUE_NAME;
Expand Down Expand Up @@ -63,7 +62,7 @@ public GetQueuesItemsCountHandler(
this.filterPattern = filterPattern;
this.queuesPrefix = queuesPrefix;
this.redisProvider = redisProvider;
this.upperBoundParallel = new UpperBoundParallel(vertx, exceptionFactory);
this.upperBoundParallel = new UpperBoundParallel(vertx);
this.exceptionFactory = exceptionFactory;
this.redisRequestQuota = redisRequestQuota;
}
Expand Down Expand Up @@ -138,10 +137,10 @@ public void handle(AsyncResult<Response> handleQueues) {
var obj = new JsonObject().put(STATUS, OK).put(QUEUES, result);
long jsonCreateDurationMs = currentTimeMillis() - beginEpchMs;
if (jsonCreateDurationMs > 10) {
log.info("Creating JSON with {} entries did block this thread for {}ms",
log.info("Creating JSON with {} entries did block this tread for {}ms",
ctx.queueLengths.length, jsonCreateDurationMs);
}else{
log.debug("Creating JSON with {} entries did block this thread for {}ms",
log.debug("Creating JSON with {} entries did block this tread for {}ms",
ctx.queueLengths.length, jsonCreateDurationMs);
}
workerPromise.complete(obj);
Expand Down
10 changes: 2 additions & 8 deletions src/main/java/org/swisspush/redisques/handler/RedisquesHttpRequestHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import org.swisspush.redisques.QueueStatsService;
import org.swisspush.redisques.QueueStatsService.GetQueueStatsMentor;
import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
import org.swisspush.redisques.performance.BurstSquasher;
import org.swisspush.redisques.util.DequeueStatistic;
import org.swisspush.redisques.util.DequeueStatisticCollector;
import org.swisspush.redisques.util.QueueStatisticsCollector;
Expand All @@ -38,7 +37,6 @@
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import static org.swisspush.redisques.util.HttpServerRequestUtil.*;
Expand Down Expand Up @@ -79,7 +77,6 @@ public class RedisquesHttpRequestHandler implements Handler<HttpServerRequest> {
private final RedisQuesExceptionFactory exceptionFactory;
private final QueueStatsService queueStatsService;
private final GetQueueStatsMentor<RoutingContext> queueStatsMentor = new MyQueueStatsMentor();
private static final AtomicReference<BurstSquasher<Object[]>> failedToServeQueueStatsLogger = new AtomicReference<>();

public static void init(
Vertx vertx, RedisquesConfiguration modConfig, QueueStatisticsCollector queueStatisticsCollector,
Expand Down Expand Up @@ -608,11 +605,8 @@ void resumeJsonWriting() {
@Override
public void onError(Throwable ex, RoutingContext ctx) {
if (!ctx.response().headWritten()) {
failedToServeQueueStatsLogger.compareAndSet(null, new BurstSquasher<>(vertx, (int count, Object[] detail) -> {
log.debug("TODO error handling {}", detail[0], exceptionFactory.newException(
"Failed to serve queue stats "+ count +" times", (Throwable)detail[1]));
}));
failedToServeQueueStatsLogger.get().logSomewhen(new Object[]{ctx.request().uri(), ex});
log.debug("TODO error handling {}", ctx.request().uri(), exceptionFactory.newException(
"Failed to serve queue stats", ex));
StatusCode rspCode = tryExtractStatusCode(ex, StatusCode.INTERNAL_SERVER_ERROR);
respondWith(rspCode, ex.getMessage(), ctx.request());
} else {
Expand Down
122 changes: 0 additions & 122 deletions src/main/java/org/swisspush/redisques/performance/BurstSquasher.java
View file

This file was deleted.

Loading

0 comments on commit fe125e2

Please sign in to comment.