This closes When a Provenance Query is submitted to WriteAheadProvenanceRepository, purge any obsolete queries from the internal state before rejecting the query due to 'too many outstanding queries'
@@ -154,7 +154,7 @@ public class LuceneEventIndex implements EventIndex {
maintenanceExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("Provenance Repository Maintenance"));
maintenanceExecutor.scheduleWithFixedDelay(() -> performMaintenance(), 1, 1, TimeUnit.MINUTES);
- maintenanceExecutor.scheduleWithFixedDelay(new RemoveExpiredQueryResults(), 30, 30, TimeUnit.SECONDS);
+ maintenanceExecutor.scheduleWithFixedDelay(this::purgeObsoleteQueries, 30, 30, TimeUnit.SECONDS);
cachedQueries.add(new LatestEventsQuery());
cachedQueries.add(new LatestEventsPerProcessorQuery());
@@ -633,9 +633,12 @@ public class LuceneEventIndex implements EventIndex {
private void validate(final Query query) {
final int numQueries = querySubmissionMap.size();
if (numQueries > MAX_UNDELETED_QUERY_RESULTS) {
+ purgeObsoleteQueries();
+ if (querySubmissionMap.size() > MAX_UNDELETED_QUERY_RESULTS) {
throw new IllegalStateException("Cannot process query because there are currently " + numQueries + " queries whose results have not "
+ "been deleted due to poorly behaving clients not issuing DELETE requests. Please try again later.");
}
+ }
if (query.getEndDate() != null && query.getStartDate() != null && query.getStartDate().getTime() > query.getEndDate().getTime()) {
throw new IllegalArgumentException("Query End Time cannot be before Query Start Time");
@@ -702,9 +705,7 @@ public class LuceneEventIndex implements EventIndex {
return removed;
}
- private class RemoveExpiredQueryResults implements Runnable {
- @Override
- public void run() {
+ private void purgeObsoleteQueries() {
try {
final Date now = new Date();
@@ -733,4 +734,3 @@ public class LuceneEventIndex implements EventIndex {
}
}
}
-}
