How to publish single event from multiple services

Question

Let's say I have a system that parses network traffic, and generates event of things that happened. example:

• ip1 was talking to ip2
• ip2 was talking to ip3
• etc

Now let's say that these events are published into event sourcing thing (like kafka for example).

I have another service (service B) that listens to these events, and keeps track of all IPs in the system. so that the work of service B is like:

ip = read_ip_from_event()
if not ip_exists(ip):
    add_new_ip(ip)

add_new_ip also publishes event (IpAddedEvent) in kafka.

Now here is my question: Let's say that I have a lot of traffic and I want to have multiple instances of service B. how can I make sure that eventually I get only one IpAddedEvent?

An issue can happen when ip doesn't exist, and two instances of serviceB reads from kafka (two different events with same ip), and both check ip_exists at the same time, and then both publish IpAddedEvent.

How can I avoid that, publishing only a single event?

Answer 1

The problem can be solved in different ways:

1. When you call ip_exists(ip) you are essentially checking some data store for an IP. This data store should be consistent and distributed and should be accessible amongst multiple instances of service B to ensure that each call to ip_exists() will return the same result regardless of the instance.

2. Your Kafka messages partitioning logic should ensure that only a subset of messages, perhaps messages within a given IP range go to one partition and within another IP range should land in another partition. This ensures that multiple instances of service B will partake of only a subset of IPs. This logic should be implemented at the producer side.

Answer 2

Since the Community Bot dug out this corpse, I'll try an alternative answer:

To ensure at most one IpAddedEvent it's actually easiest to have a single instance of service B. If you're bothered that this single instance would be a bottleneck, you can reduce its input feed frequency by pre-filtering.

The pre-filter processes would either be part of the original traffic sniffers, or they would listen to the message events created by the traffic sniffers, and keep a relatively small in-memory cache of recently seen IP addresses. Only for new IP addresses they generate an IpSeenEvent (for example through a private topic) that is processed by service B which then utilizes its long-term memory by virtue of ip_exists() to ensure that it only posts IpAddedEvent for actually new IP addresses. Since truly new IP addresses should be relatively rare, doing this in a single service worker should work fairly well.