joerghoh commented on code in PR #184:
URL:
https://github.com/apache/sling-org-apache-sling-distribution-journal/pull/184#discussion_r2918701947
##########
src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java:
##########
@@ -209,6 +226,10 @@ public OffsetQueue<DistributionQueueItem>
getOffsetQueue(String pubAgentName, lo
@Override
public int getMaxQueueSize(String pubAgentName) {
+ return cachedQueueSizes.computeIfAbsent(pubAgentName,
this::computeMaxQueueSize);
Review Comment:
that means the if the calcuation in the executor has not yet finished, we
try here in a synchronous way as well (that means that the calculation is now
running twice). Is that really necessary? Can we just return -1 until the
initial calculation has been completed?
##########
src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java:
##########
@@ -77,13 +82,24 @@ public class PubQueueProviderImpl implements
PubQueueProvider, Runnable {
*/
private final Map<String, OffsetQueue<Long>> errorQueues = new
ConcurrentHashMap<>();
+ private final Map<String, Integer> cachedQueueSizes = new
ConcurrentHashMap<>();
+
+ private final ScheduledExecutorService queueSizeExecutor;
+
private ServiceRegistration<?> reg;
public PubQueueProviderImpl(EventAdmin eventAdmin, QueueErrors
queueErrors, CacheCallback callback, BundleContext context) {
queuedNotifier = new PackageQueuedNotifier(eventAdmin);
this.queueErrors = queueErrors;
this.callback = callback;
cache = newCache();
+ queueSizeExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "queue-size-refresh");
+ t.setDaemon(true);
+ return t;
+ });
+ queueSizeExecutor.scheduleAtFixedRate(this::refreshQueueSizes,
+ QUEUE_SIZE_REFRESH_SECONDS, QUEUE_SIZE_REFRESH_SECONDS,
TimeUnit.SECONDS);
Review Comment:
Why do you delay the first computation of the queue by
``QUEUE_SIZE_REFRESH_SECONDS``? Can't it start right away?
(besides, the double usage of ``QUEUE_SIZE_REFRESH_SECONDS`` for 2 different
purposes is a bit misleading.)
##########
src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java:
##########
@@ -99,6 +115,7 @@ private void startCleanupTask(BundleContext context) {
@Override
public void close() {
+ queueSizeExecutor.shutdownNow();
Review Comment:
this will trigger ``Thread.interrupt()`` on all threads currently active in
the Executor; and if that thread has the chance of reading data from Apache
Oak, it is likely to blow the repository, and we should avoid that ...
However, if we can rule out that these threads ever reach into the
repository, we can use it.
##########
src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java:
##########
@@ -77,13 +82,24 @@ public class PubQueueProviderImpl implements
PubQueueProvider, Runnable {
*/
private final Map<String, OffsetQueue<Long>> errorQueues = new
ConcurrentHashMap<>();
+ private final Map<String, Integer> cachedQueueSizes = new
ConcurrentHashMap<>();
+
+ private final ScheduledExecutorService queueSizeExecutor;
+
private ServiceRegistration<?> reg;
public PubQueueProviderImpl(EventAdmin eventAdmin, QueueErrors
queueErrors, CacheCallback callback, BundleContext context) {
queuedNotifier = new PackageQueuedNotifier(eventAdmin);
this.queueErrors = queueErrors;
this.callback = callback;
cache = newCache();
+ queueSizeExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "queue-size-refresh");
+ t.setDaemon(true);
+ return t;
+ });
+ queueSizeExecutor.scheduleAtFixedRate(this::refreshQueueSizes,
+ QUEUE_SIZE_REFRESH_SECONDS, QUEUE_SIZE_REFRESH_SECONDS,
TimeUnit.SECONDS);
Review Comment:
from what I see this changes the behavior in a way, that the new queue size
is computed every 30 seconds. Feels a bit high for me, as I can imagine
situations where a quicker feedback would be helpful ... can we make that value
configurable (the default of 30 is ok)
##########
src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java:
##########
@@ -217,6 +238,16 @@ public int getMaxQueueSize(String pubAgentName) {
}
}
+ private void refreshQueueSizes() {
+ for (String agentName : cachedQueueSizes.keySet()) {
+ try {
+ cachedQueueSizes.put(agentName,
computeMaxQueueSize(agentName));
Review Comment:
what is the expected runtime of ``computeMaxQueueSize()``? If that scales
linearly with the number of entries, we should have timing metrics for this
calculation, because then it will increase the delay between updates of the
resulting numbers from 30 seconds to an potentially even higher value. At some
point (for example at 60 seconds) I would start logging INFO or WARN messages,
that the calculation took really long.
(And that also means, that the staleness increases ... Not that it was any
better before, but now it gets explicit. And we should think about documenting
it properly.)
##########
src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java:
##########
@@ -62,6 +65,8 @@ public class PubQueueProviderImpl implements
PubQueueProvider, Runnable {
*/
private static final int CLEANUP_THRESHOLD = 10_000;
+ static final long QUEUE_SIZE_REFRESH_SECONDS = 30;
Review Comment:
```suggestion
static final long QUEUE_SIZE_REFRESH_INTERVAL_SECONDS = 30;
```
Now it is clearly an interval, not just a number.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
