Conversation
Reviewer's Guide by SourceryThis pull request introduces several enhancements to the event processing pipeline. It ensures events are processed in the correct order by verifying the existence of previous events, prevents concurrent catch-up operations for the same topic, and improves logging for better traceability. Additionally, it ensures that the HWM is initialized only once and events are published only after a successful HWM update. Sequence diagram for Event Processing with HWM ChecksequenceDiagram
participant OP as OrderedProcessor
participant DB as Database
participant MB as MessageBroker
OP->DB: Check for unprocessed prior events
DB-->OP: Returns if prior events exist
alt Prior events exist
OP->OP: Skip event
else No prior events
OP->DB: Update event status to 'p'
DB-->OP: Returns update status
alt Update failed
OP->OP: Skip event
else Update successful
OP->DB: Check if previous event exists (HWM check)
DB-->OP: Returns if previous event exists
alt Previous event does not exist
OP->OP: Ignore event
else Previous event exists
OP->DB: Commit transaction
DB-->OP: Acknowledge commit
OP->MB: Publish event
MB->MB: Deliver to subscribers
end
end
end
Updated class diagram for EventclassDiagram
class Event {
String id
String source
String type
String datacontenttype
String dataschema
String subject
byte[] data
Instant time
Long idn
String topic
+create(String id, String source, String type, String datacontenttype, String dataschema, String subject, byte[] data) Event
+create(String id, String source, String type, String datacontenttype, String dataschema, String subject, byte[] data, Instant time, Long idn, String topic) Event
}
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey @p14n - I've reviewed your changes - here's some feedback:
Overall Comments:
- Consider using a dedicated thread pool for asynchronous tasks to better manage resources.
- The logging could be improved by including more context, such as the topic name.
Here's what I looked at during the review
- 🟡 General issues: 5 issues found
- 🟢 Security: all looks good
- 🟢 Testing: all looks good
- 🟡 Complexity: 1 issue found
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| try (ResultSet rs = stmt.executeQuery()) { | ||
| if (rs.next()) { | ||
| int hwm = rs.getInt(1); | ||
| System.err.println("HWM: " + hwm + " IDN: " + event.idn()); |
There was a problem hiding this comment.
suggestion: Avoid using System.err.println for logging in production code.
Using the project's Logger would ensure consistent log formatting and levels. Consider replacing System.err.println with a logger call.
| System.err.println("HWM: " + hwm + " IDN: " + event.idn()); | |
| LOGGER.debug("HWM: {} IDN: {}", hwm, event.idn()); |
| public void publish(InT message) { | ||
|
|
||
| if(!canProcess(message)){ | ||
| System.err.println("DMB got event " + message); |
There was a problem hiding this comment.
suggestion: Use the configured Logger instead of System.err.println.
Utilizing the Logger facilitates centralized log management and better control over logging levels. This change will enhance maintainability.
Suggested implementation:
package com.p14n.postevent.broker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class DefaultMessageBroker<InT, OutT> {
private static final Logger logger = LoggerFactory.getLogger(DefaultMessageBroker.class); @Override
public void publish(InT message) {
logger.info("DMB got event {}", message);
if (!canProcess(message)) {
return;
}
// Deliver to all subscribers
for (MessageSubscriber<OutT> subscriber : subscribers) {
logger.info("TO ASYNC {}", message);
asyncExecutor.submit(() -> {
try {
subscriber.onMessage(convert(message));| signals.set(0); | ||
| catchup(message.topic); | ||
| running.set(false); | ||
| if (signals.get() > 0) { | ||
| onMessage(message); | ||
| } |
There was a problem hiding this comment.
suggestion (performance): Recursion in onMessage may risk stack overflows.
Re-invoking onMessage recursively when signals are pending could lead to deep recursion under heavy load. Consider an iterative approach or scheduling a new execution to handle pending signals.
| signals.set(0); | |
| catchup(message.topic); | |
| running.set(false); | |
| if (signals.get() > 0) { | |
| onMessage(message); | |
| } | |
| do { | |
| signals.set(0); | |
| catchup(message.topic); | |
| } while (signals.get() > 0); | |
| running.set(false); |
|
|
||
| // Forward to actual subscriber after successful persistence | ||
| targetBroker.publish(event); | ||
| System.err.println("PB got event " + event.id()); |
There was a problem hiding this comment.
suggestion: Replace System.err.println with proper logging.
To maintain logging consistency and log level management across modules, it is advisable to use the established Logger instead of System.err.println.
Suggested implementation:
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// other imports...public class PersistentBroker {
private static final Logger LOGGER = LoggerFactory.getLogger(PersistentBroker.class); LOGGER.info("PB got event {}", event.id());If PersistentBroker already has a logger instance declared, remove the duplicate declaration. Also, adjust the logger level if info is not appropriate.
| Consumer<ChangeEvent<String, String>> consumer = record -> { | ||
| try { | ||
| Event event = changeEventToEvent(record); | ||
| System.err.println("LC got event " + event.id()); |
There was a problem hiding this comment.
suggestion: Utilize the Logger rather than System.err.println for client notifications.
Switching to the Logger will keep logging behavior consistent across components and allow for better control over output, especially in production environments.
Suggested implementation:
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// (existing imports)public class LocalConsumer {
private static final Logger logger = LoggerFactory.getLogger(LocalConsumer.class); logger.info("LC got event " + event.id());If the project uses a different logging framework (e.g., Log4j2) adjust the imports and Logger instantiation accordingly. Ensure that the Logger configuration is set properly in your application configuration.
| private AtomicInteger signals = new AtomicInteger(0); | ||
| private AtomicBoolean running = new AtomicBoolean(false); | ||
|
|
||
| @Override |
There was a problem hiding this comment.
issue (complexity): Consider using a loop with a try-finally block to process pending signals, which avoids recursion and reduces lock contention by using a dedicated lock and ensuring the running flag is reset even if an exception occurs.
The recursive call and mix of atomic flags can be simplified by processing pending signals in a single loop, preventing recursion and reducing lock contention. One option is to use a loop within a try-finally block to drain the queued signals. For example:
@Override
public void onMessage(SystemEvent message) {
if (Objects.requireNonNull(message) == SystemEvent.CatchupRequired) {
signals.incrementAndGet();
synchronized (this) { // Using 'this' as the lock instead of running
if (running.get()) {
return;
}
running.set(true);
}
try {
while (signals.getAndSet(0) > 0) {
catchup(message.topic);
}
} finally {
running.set(false);
}
}
}Actionable Steps:
- Replace the recursive call with a
whileloop to process any queued catchup signals. - Use a dedicated lock (e.g., synchronizing on
this) to ensure mutual exclusion. - Use a
try-finallyblock to guarantee that the running flag is reset even if an exception occurs.
This keeps all functionality intact while reducing complexity and potential concurrency pitfalls.
1 participant
Addresses #33
Summary by Sourcery
Augment the distributed system's event processing capabilities with improved event ordering, tracking, and synchronization mechanisms
New Features:
Enhancements:
Build:
Tests: