Resubmission of publications must not unregister in-progress publications in case of successful invocation #1051
Description
The processIncompletePublications method includes a retry mechanism for failed events and leverages the inProgress cache to improve efficiency. However, there is an issue:
1.In the finally block, the inProgress.unregister(it) method is redundantly called.
2.This leads to unnecessary cache invalidation because consuming events might be an asynchronous operation that has not been completed.
3.Both the markFailed and markCompleted methods already handle the inProgress.unregister operation, making the finally block unnecessary.
Suggested Fix:
Remove the finally block containing inProgress.unregister(it) to avoid redundant operations and improve code clarity.
Code Example of Current Implementation:
publications.stream() //
.filter(filter) //
.forEach(it -> {
try {
inProgress.register(it);
consumer.accept(it);
} catch (Exception o_O) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Error republishing event publication %s.".formatted(it), o_O);
}
} finally {
inProgress.unregister(it);
}
});
Proposed Fix:
publications.stream() //
.filter(filter) //
.forEach(it -> {
try {
inProgress.register(it);
consumer.accept(it);
} catch (Exception o_O) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Error republishing event publication %s.".formatted(it), o_O);
}
}
});