Events notification system for Nessie - Reference Implementation #6943

adutra · 2023-05-31T09:54:27Z

No description provided.

adutra · 2023-05-31T10:00:29Z

@omarsmak and @tomekl007 : adding you both as reviewers, since you showed interest in this :-)

tomekl007

Awesome work!

tomekl007 · 2023-05-31T10:05:18Z

events/ri/README.md

+
+### Serialization and deserialization
+
+`KafkaEventSubscriber` uses the [Avro] library to define a domain model that is specific to the subscriber, and is not


That's great that we use Avro.

tomekl007 · 2023-05-31T10:11:28Z

events/ri/README.md

+
+In this example, the destination Kafka topic is going to receive messages of three different types. This is a common
+scenario in Kafka, but requires some extra work to be done on both the producer and consumer sides, especially with
+respect to schema handling. The strategy we chose to use here is outlined in [Martin Kleppmann's excellent article] and


Can you elaborate more on the most important aspect in our context of why we picked this strategy?
Was it any events that need to stay in a fixed order must go in the same topic

I chose this strategy for the reference impl because this strategy does not require to pre-configure the schema registry, which makes it easier to test the producer.

IOW, when the topic is created and the first message comes in, the producer will upload the schema since it's not present in the registry yet. It will do this for each new message type, since the strategy is to have one schema per (topic + message type) pair.

Other strategies, for example those outlined in Robert Yokota's article, require the registry to be set up with all the schemas prior to using the topic.

tomekl007 · 2023-05-31T10:14:10Z

events/ri/README.md

+### Using headers efficiently
+
+The `KafkaEventSubscriber` implementation also illustrates how to use Kafka headers to pass additional information along
+with the message. In this case, the subscriber adds the repository id, the user and the event creation timestamp to the


Did we consider putting creation timestamp in the event itself? Instead of headers? It maybe beneficial in the situation when the end user would like to store the events offline for further analysis. If the creation timestamp is not there, users will need to create a new avro schema that contains the timestamp and map our event + timestamp to this new schema and finally save the event.

Yes, the event timestamp can be part of the message if you prefer. Here I'm using headers, but it's just an example.

tomekl007 · 2023-05-31T10:15:24Z

events/ri/README.md

+
+The `KafkaEventSubscriber` implementation also illustrates how to handle events in a non-blocking way, by publishing
+messages without waiting for the broker's acknowledgement. This is achieved by using the `KafkaProducer#send` method in
+a fire-and-forget mode, along with the `acks=0` producer configuration option. If your implementation needs to wait for


As far as I know, using acks=0 is not recommended in the majority of cases. I am not sure if using 0 in the reference implementation is the best solution.

Sorry for the late reply. You are right, and I had to open my Kafka in Action book to refresh my skills around acks and send methods :-)

In short, we have 3 send methods:

fire-and-forget: don't wait for the future to complete, don't provide any callback;

async: don't wait for the future to complete, but provide a completion callback;

sync: wait for the future to complete (synchronously).

With Nessie, 1 and 2 are fine, but 3 should be avoided (or at least, isBlocking() should return true).

The way these methods really work will further depend on the acks setting (and retries). In short, the returned future is completed when:

With acks=0 => immediately when the message is sent;

With acks=1 => when the broker acknowledged the write;

With acks=all => when the broker and all the in-sync replicas acknowledged the write.

Which brings us to the conclusion: the number of acks is relatively orthogonal to the send method.

With all that in mind, I will indeed change the example to use the async send method AND ack=1, since both seem reasonable choices for this use case.

tomekl007 · 2023-05-31T10:18:41Z

events/ri/src/main/avro/NessieEvents.avdl

@@ -0,0 +1,39 @@
+@namespace("com.example.nessie.events.generated")
+protocol NessieEvents {


Does it model all the possible events or only a subset of them?
If the latter, maybe we should document for end users what's the best way to extend it.

It doesn't model merge and transplant events indeed, since in this example, we don't process them. I will add a note.

Done. I added comments to all record types in the Avro schema, and also included a new record type to model Merges and Transplants (it's not used in the example, but the record definition is there). Let me know what you think.

tomekl007 · 2023-05-31T10:21:00Z

events/ri/src/main/java/org/projectnessie/events/ri/kafka/KafkaEventSubscriber.java

+  public boolean isBlocking() {
+    // We don't wait for the broker acknowledgement, so we can tell Nessie
+    // that we don't do blocking I/O.
+    return false;


how does the nessie handle the situation when isBlocking==true?

It defers the call to onXYZ to a worker thread, as opposed to calling the method on the event loop thread.

tomekl007 · 2023-05-31T10:23:25Z

events/ri/src/main/java/org/projectnessie/events/ri/kafka/KafkaEventSubscriber.java

+    String repositoryId = upstreamEvent.getRepositoryId();
+    String referenceName;
+    if (downstreamEvent instanceof CommitEvent) {
+      referenceName = ((CommitEvent) downstreamEvent).getTargetReference();


Why the naming inconsistency? getTargetReference vs getReference below.

In Nessie's Events API, a CommitEvent inherits from CommittingEvent where 2 methods are defined: getSourceReference() and getTargetReference(). In the case of a commit, both references are always the same (but they are different for merges and transplants).

Whereas a ReferenceEvent has a distinct getReference() method, because it's always about one single reference.

This was recently changed FYI, a CommitEvent now only has one getReference() method.

tomekl007 · 2023-05-31T10:24:04Z

events/ri/src/main/resources/META-INF/services/org.projectnessie.events.spi.EventSubscriber

+org.projectnessie.events.ri.console.PrintingEventSubscriber
+
+# Kafka event subscriber
+org.projectnessie.events.ri.kafka.KafkaEventSubscriber


Is there a limit on the number of Subscribers that we can support?

No hard limit, but the more subscribers, the more you mobilize Nessie's event loop and/or worker threads. I am planning some benchmarks later on.

tomekl007 · 2023-05-31T10:24:54Z

events/ri/src/intTest/java/org/projectnessie/events/ri/kafka/ITKafkaEventSubscriber.java

+        AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
+        "http://localhost:" + SCHEMA_REGISTRY.getMappedPort(8081));
+    props.put(
+        AbstractKafkaSchemaSerDeConfig.VALUE_SUBJECT_NAME_STRATEGY, TopicRecordNameStrategy.class);


Where do we specify what part (which field) of the record is used by the TopicRecordNameStrategy?

In this strategy, the schema is recorded under the subject <topic name>-<schema name>, so you don't need to select a field in the record. E.g. for the Avro ReferenceEvent message, the schema name is ReferenceEvent.

omarsmak

Question, I guess we can't have an example based on KafkaConnect since the producer runs in Nessie server, correct?

omarsmak · 2023-06-01T13:40:03Z

events/ri/README.md

+### Serialization and deserialization
+
+`KafkaEventSubscriber` uses the [Avro] library to define a domain model that is specific to the subscriber, and is not
+tied in any way to Nessie's. This is indeed the recommended approach, as it allows the subscriber to define its own


I am not sure why tiding the implementation with Avro? Wouldn't make sense to have approach to let users definer their "choice" format? You could have some prefer using Avro, some may prefer protobuf and some may even prefer JSON :) . However, since is an example, could be fine though.

I think this would go beyond the purpose of a reference implementation. As you said, this is an example and what matters most imho is that people can easily adapt it to their own needs. Here, it shouldn't be very hard to switch to protobuf for instance, so I don't see a lot of value in actually providing the protobuf bindings + another subscriber.

adutra · 2023-06-01T17:43:03Z

I guess we can't have an example based on KafkaConnect since the producer runs in Nessie server, correct?

That, and also the fact that providing a full-blown source connector for Nessie would certainly be an interesting project to do, but also a bit out of scope for a reference implementation of the Events SPI module (especially because it would be less portable / adaptable by a variety of users).

adutra · 2023-06-06T08:31:56Z

@tomekl007 or @omarsmak could I get another round of review please?

snazy · 2023-06-06T13:17:06Z

events/ri/build.gradle.kts

+
+plugins {
+  id("nessie-conventions-server")
+  id("com.github.davidmc24.gradle.plugin.avro") version "1.7.1"


Should be defined in libs.versions.toml

snazy · 2023-06-06T13:17:18Z

gradle/projects.main.properties

@@ -6,6 +6,7 @@ nessie-compatibility-tests=compatibility/compatibility-tests
 nessie-compatibility-jersey=compatibility/jersey
 nessie-events-api=events/api
 nessie-events-quarkus=events/quarkus
+nessie-events-ri=events/ri


Not in the nessie-bom?

None of the event modules is declared in :nessie-bom... I guess that's an oversight.

This commit introduces a reference implementation for the Nessie events notification system. The reference implementation showcases an Apache Kafka producer putting events from Nessie in a Kafka topic, using Apache Avro for the messages.

adutra requested review from omarsmak and tomekl007 May 31, 2023 09:54

adutra force-pushed the events-ri branch from c01061b to 22b7ea3 Compare May 31, 2023 10:09

tomekl007 reviewed May 31, 2023

View reviewed changes

snazy added this to the 0.61.0 milestone May 31, 2023

adutra mentioned this pull request May 31, 2023

Enhancements to the Events API #6945

Merged

omarsmak reviewed Jun 1, 2023

View reviewed changes

adutra force-pushed the events-ri branch from dc1f806 to a8ed83b Compare June 1, 2023 14:43

snazy reviewed Jun 6, 2023

View reviewed changes

snazy modified the milestones: 0.61.0, 0.62.0 Jun 12, 2023

snazy requested review from omarsmak and tomekl007 June 14, 2023 14:25

snazy modified the milestones: 0.62.0, 0.63.0, 0.64.0 Jun 14, 2023

snazy modified the milestones: 0.64.0, 0.65.0, 0.66.0 Jun 27, 2023

snazy removed this from the 0.66.0 milestone Jul 6, 2023

adutra force-pushed the events-ri branch 2 times, most recently from 9fe6a3e to 6409007 Compare August 4, 2023 09:30

adutra force-pushed the events-ri branch from 6409007 to 6e21abb Compare October 10, 2023 13:22

adutra force-pushed the events-ri branch from 6e21abb to 83f51fd Compare April 17, 2024 14:43

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Events notification system for Nessie - Reference Implementation #6943

Events notification system for Nessie - Reference Implementation #6943

adutra commented May 31, 2023

adutra commented May 31, 2023

tomekl007 left a comment

tomekl007 May 31, 2023 •

edited

tomekl007 May 31, 2023

adutra May 31, 2023 •

edited

tomekl007 May 31, 2023

adutra May 31, 2023

tomekl007 May 31, 2023

adutra Jun 2, 2023

tomekl007 May 31, 2023

adutra May 31, 2023

adutra Jun 2, 2023

tomekl007 May 31, 2023

adutra May 31, 2023

tomekl007 May 31, 2023

adutra May 31, 2023

adutra Jun 1, 2023

tomekl007 May 31, 2023

adutra May 31, 2023

tomekl007 May 31, 2023

adutra May 31, 2023

omarsmak left a comment

omarsmak Jun 1, 2023

adutra Jun 1, 2023

adutra commented Jun 1, 2023

adutra commented Jun 6, 2023

snazy Jun 6, 2023

snazy Jun 6, 2023

adutra Jun 6, 2023


		### Serialization and deserialization

		`KafkaEventSubscriber` uses the [Avro] library to define a domain model that is specific to the subscriber, and is not

		@@ -0,0 +1,39 @@
		@namespace("com.example.nessie.events.generated")
		protocol NessieEvents {

Events notification system for Nessie - Reference Implementation #6943

Are you sure you want to change the base?

Events notification system for Nessie - Reference Implementation #6943

Conversation

adutra commented May 31, 2023

adutra commented May 31, 2023

tomekl007 left a comment

Choose a reason for hiding this comment

tomekl007 May 31, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adutra May 31, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

omarsmak left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adutra commented Jun 1, 2023

adutra commented Jun 6, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tomekl007 May 31, 2023 •

edited

adutra May 31, 2023 •

edited