Conversation
Contributor
Reviewer's Guide by SourceryThis pull request introduces support for multiple topics in the post-event system. The changes include modifications to the configuration, broker, consumer, and database setup to handle multiple topics. The pull request also includes a new test case to verify the functionality of multiple topics with dedicated clients. Updated ER diagram for postevent.messages tableerDiagram
messages {
bigint idn PK
VARCHAR topic PK
VARCHAR id
VARCHAR source
VARCHAR data_schema
VARCHAR subject
bytea data
TIMESTAMP time
VARCHAR status
}
Updated class diagram for DefaultMessageBrokerclassDiagram
class DefaultMessageBroker~InT, OutT~ {
-ConcurrentHashMap~String, Set~MessageSubscriber~OutT~~~ topicSubscribers
-AtomicBoolean closed
-AsyncExecutor asyncExecutor
+DefaultMessageBroker()
+DefaultMessageBroker(AsyncExecutor asyncExecutor)
-canProcess(String topic, InT message) bool
+publish(String topic, InT message) void
+subscribe(String topic, MessageSubscriber~OutT~ subscriber) bool
+unsubscribe(String topic, MessageSubscriber~OutT~ subscriber) bool
+close() void
}
DefaultMessageBroker : implements MessageBroker
DefaultMessageBroker *-- AsyncExecutor : has a
DefaultMessageBroker -- MessageSubscriber : Uses
note for DefaultMessageBroker "Supports multiple topics using topicSubscribers"
Updated class diagram for MessageBrokerGrpcClientclassDiagram
class MessageBrokerGrpcClient {
-Logger logger
-MessageBrokerServiceGrpc.MessageBrokerServiceStub asyncStub
-Set~String~ subscribed
ManagedChannel channel
+MessageBrokerGrpcClient(String host, int port)
+MessageBrokerGrpcClient(ManagedChannel channel)
+subscribeToEvents(String topic) void
-convertFromGrpcEvent(EventResponse grpcEvent) Event
+publish(String topic, Event message) void
+subscribe(String topic, MessageSubscriber~Event~ subscriber) bool
+unsubscribe(String topic, MessageSubscriber~Event~ subscriber) bool
}
MessageBrokerGrpcClient -- Event : Publishes
MessageBrokerGrpcClient : implements MessageBroker
MessageBrokerGrpcClient -- MessageSubscriber : Uses
note for MessageBrokerGrpcClient "Supports multiple topics and subscriptions"
Updated class diagram for ConsumerClientclassDiagram
class ConsumerClient {
-Logger logger
-AsyncExecutor asyncExecutor
-List~AutoCloseable~ closeables
-TransactionalBroker tb
SystemEventBroker seb
+ConsumerClient(AsyncExecutor asyncExecutor)
+ConsumerClient()
+start(Set~String~ topics, DataSource ds, String host, int port) void
+start(Set~String~ topics, DataSource ds, ManagedChannel channel) void
+close() void
+publish(String topic, TransactionalEvent message) void
+subscribe(String topic, MessageSubscriber~TransactionalEvent~ subscriber) bool
+unsubscribe(String topic, MessageSubscriber~TransactionalEvent~ subscriber) bool
+convert(TransactionalEvent m) TransactionalEvent
}
ConsumerClient -- TransactionalEvent : Publishes
ConsumerClient : implements MessageBroker
ConsumerClient -- MessageSubscriber : Uses
note for ConsumerClient "Supports multiple topics and subscriptions"
Updated class diagram for PersistentBrokerclassDiagram
class PersistentBroker~OutT~ {
-Logger logger
-String INSERT_SQL
-String UPDATE_HWM_SQL
-MessageBroker~Event, OutT~ targetBroker
-DataSource dataSource
-AsyncExecutor asyncExecutor
+PersistentBroker(MessageBroker~Event, OutT~ targetBroker, DataSource dataSource, AsyncExecutor asyncExecutor)
+publish(String topic, Event event) void
+subscribe(String topic, MessageSubscriber~OutT~ subscriber) bool
+unsubscribe(String topic, MessageSubscriber~OutT~ subscriber) bool
+close() void
+convert(Event m) OutT
+onMessage(Event message) void
}
PersistentBroker : implements MessageBroker
PersistentBroker : implements MessageSubscriber
PersistentBroker -- Event : Publishes
PersistentBroker -- MessageSubscriber : Uses
note for PersistentBroker "Supports multiple topics and subscriptions"
Updated class diagram for LocalConsumerclassDiagram
class LocalConsumer~OutT~ {
-Logger logger
-DebeziumServer debezium
-MessageBroker~Event, OutT~ broker
-PostEventConfig config
-DatabaseSetup db
+LocalConsumer(PostEventConfig config, MessageBroker~Event, OutT~ broker)
+start() void
+stop() void
+close() void
}
LocalConsumer -- Event : Publishes
LocalConsumer -- MessageBroker : Uses
note for LocalConsumer "Supports multiple topics"
Updated class diagram for OrderedProcessorclassDiagram
class OrderedProcessor {
-Logger logger
-BiFunction~Connection, Event, Boolean~ processorFunction
+OrderedProcessor(BiFunction~Connection, Event, Boolean~ processorFunction)
+process(Connection connection, Event event) bool
-processEventWithFunction(Connection connection, Event event) bool
-hasUnprocessedPriorEvents(Connection connection, Event event) bool
-previousEventExists(Connection connection, Event event) bool
-updateEventStatus(Connection connection, Event event) bool
}
OrderedProcessor -- Event : Processes
note for OrderedProcessor "Filters events by topic"
Updated class diagram for TransactionalBrokerclassDiagram
class TransactionalBroker {
-Logger logger
-DataSource ds
+TransactionalBroker(DataSource ds)
+publish(String topic, Event message) void
+subscribe(String topic, MessageSubscriber~TransactionalEvent~ subscriber) bool
+unsubscribe(String topic, MessageSubscriber~TransactionalEvent~ subscriber) bool
+convert(Event m) TransactionalEvent
}
TransactionalBroker -- Event : Publishes
TransactionalBroker -- MessageSubscriber : Uses
note for TransactionalBroker "Supports multiple topics and subscriptions"
Updated class diagram for SystemEventBrokerclassDiagram
class SystemEventBroker {
+convert(SystemEvent m) SystemEvent
+publish(SystemEvent event) void
+subscribe(MessageSubscriber~SystemEvent~ subscriber) void
}
SystemEventBroker -- SystemEvent : Publishes
SystemEventBroker -- MessageSubscriber : Uses
note for SystemEventBroker "Supports system events"
Updated class diagram for PostEventConfigclassDiagram
class PostEventConfig {
+affinity() String
+topics() Set~String~
+dbHost() String
+dbPort() int
+dbUser() String
+dbPassword() String
+dbName() String
+overrideProps() Properties
+startupTimeoutSeconds() int
+jdbcUrl() String
}
note for PostEventConfig "Supports multiple topics"
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}
Contributor
There was a problem hiding this comment.
Hey @p14n - I've reviewed your changes - here's some feedback:
Overall Comments:
- The changes to DefaultMessageBroker look good, but it would be good to add a test case that verifies that messages are only delivered to subscribers of the correct topic.
- The new test case
testMultipleTopicsWithDedicatedClientsis a great addition to verify the multi-topic functionality.
Here's what I looked at during the review
- 🟡 General issues: 1 issue found
- 🟢 Security: all looks good
- 🟡 Testing: 1 issue found
- 🟡 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.
| subscribed.set(false); | ||
| public boolean unsubscribe(String topic, MessageSubscriber<Event> subscriber) { | ||
| boolean unsubscribed = super.unsubscribe(topic, subscriber); | ||
| if (topicSubscribers.isEmpty()) { |
Contributor
There was a problem hiding this comment.
issue (bug_risk): Check reference to undefined 'topicSubscribers'.
The unsubscribe method uses 'if (topicSubscribers.isEmpty())' but this field is not declared in MessageBrokerGrpcClient. It appears to have been carried over from another class. Likely, it should instead check whether there are any active subscriptions for the given topic (for example, via the 'subscribed' set) before shutting down the channel.
| @@ -115,6 +108,13 @@ void testDeterministicEventDelivery(@ForAll("randomSeeds") long seed) throws Exc | |||
| } | |||
| Thread.sleep(2000); | |||
Contributor
There was a problem hiding this comment.
suggestion (testing): Add a test for unsubscribing.
It would be beneficial to add a test case that specifically covers the unsubscribe functionality to ensure that clients can correctly unsubscribe from topics and no longer receive messages after doing so.
Suggested implementation:
// (Existing test methods and code)
// New test method to verify the unsubscribe functionality
@Test
public void testUnsubscribe() throws Exception {
// Setup client subscriber and capture received events
Set<Integer> receivedIds = ConcurrentHashMap.newKeySet();
// Subscribe to the topic and capture events
var subscription = client.subscribe(TOPIC, (TransactionalEvent event) -> {
receivedIds.add(event.event().idn());
});
// Allow some time for subscription to be active, and verify subscription works by waiting briefly
Thread.sleep(1000);
// Now unsubscribe to stop receiving further events
subscription.unsubscribe();
// Clear any events that might have been received prior to unsubscribing
receivedIds.clear();
// Publish a new event after unsubscribe.
// This assumes a publish method exists that sends events to TOPIC.
client.publish(TOPIC, createTestEvent());
// Allow time for event processing after publish
Thread.sleep(2000);
// Assert that no new event was received after unsubscribing
assertTrue("No events should be received after unsubscribe", receivedIds.isEmpty());
}
// Helper method to create a dummy test event.
// Implementation should be replaced with the actual parameters and logic applicable to your TransactionalEvent.
private TransactionalEvent createTestEvent() {
// TODO: Provide actual implementation details for creating a dummy event for testing
return new TransactionalEvent(/* pass necessary parameters */);
}Make sure that:
- The client.subscribe method returns an object (here named subscription) with an unsubscribe() method. If the method signature differs, adjust accordingly.
- A client.publish(TOPIC, event) method is available or replace it with the actual publish mechanism.
- All required imports are included:
- import org.junit.Test;
- import static org.junit.Assert.assertTrue;
- import java.util.Set;
- import java.util.concurrent.ConcurrentHashMap;
Also, adjust the createTestEvent() method to properly construct a valid TransactionalEvent instance based on your project’s actual implementation.
| import io.grpc.stub.StreamObserver; | ||
|
|
||
| import java.time.Instant; | ||
| import java.time.OffsetDateTime; |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider relying on the base class's tracking of per-topic subscribers instead of maintaining a separate subscribed set to simplify the logic and reduce state complexity .
Consider removing the extra subscription tracking (i.e. the separate subscribed set) and instead rely on the base class’s tracking of per‑topic subscribers. For example, if your base class maintains subscriber collections per topic you can check their size to decide when to initiate or cancel a subscription. This reduces branching and the extra state.
For instance:
@Override
public boolean subscribe(String topic, MessageSubscriber<Event> subscriber) {
if (super.subscribe(topic, subscriber)) {
// If this is the first subscriber for the topic, subscribe to events.
if (getSubscribers(topic).size() == 1) { // assuming getSubscribers(topic) is available
subscribeToEvents(topic);
}
return true;
}
return false;
}
@Override
public boolean unsubscribe(String topic, MessageSubscriber<Event> subscriber) {
boolean result = super.unsubscribe(topic, subscriber);
if (getSubscribers(topic).isEmpty()) {
shutdownTopicSubscription(topic); // implement shutdown per topic if needed
}
return result;
}This way, you remove the duplicated state, delegate the logic to the underlying topic-subscriber management, and make the code easier to follow.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Augment the message broker to support multiple topics with dedicated clients and improve event processing across different topics
New Features:
Enhancements:
Tests:
Chores: