Introduce broadcast API to TransportProvider #903

surajkn · 2022-04-17T22:26:38Z

This API will allow to sending events to all consumers/clients of TransportProvider.
TransportProvider will be able to optionally implement broadcast API.
Example, for Kafka TransportProvider broadcast will send the event to all topic partitions.

Important: DO NOT REPORT SECURITY ISSUES DIRECTLY ON GITHUB.
For reporting security issues and contributing security fixes,
please, email [email protected] instead, as described in
the contribution guidelines.

Please, take a minute to review the contribution guidelines at:
https://github.com/linkedin/Brooklin/blob/master/CONTRIBUTING.md

This API will allow to sending events to all consumers/clients of TrasnsportProvider. TransportProvider will be able to optionally implement broadcast API. Example, for Kafka TrasnportProvider broadcast will send the event to all topic partitions.

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTestUtils.java

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProvider.java

Broadcast is a best effort operation. In Broadcast we will only attempt to send to all partitions but not wait for each send to complete. Only if the attempt to send fails will an exception be passed to the callback. Broadcast callback will also be passed a metadata object with total partition count and partitions to which send attempt succeeded. TP broadcast API will also accept a second callback to be called when send to each partition completes. This will be used by EventProducer to pass a callback to update checkpoint and publish metrics. Currently EventProducer does not expose this second callback, since given the best effort nature of broadcast API it is not necessary to update outcome of send to each partition.

datastream-server-api/src/main/java/com/linkedin/datastream/server/DatastreamEventProducer.java

...server-api/src/main/java/com/linkedin/datastream/server/api/transport/TransportProvider.java

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTestUtils.java

ryannedolan

Think there are some bugs wrt exceptions and callbacks.

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProvider.java

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProvider.java

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

Broadcast method modified to accept only one callback to be called at each event send completion. Broadcast method will return DatastreamRecordMetadata containing broadcast relevant information like total partition count, partitions to which send was attempted, if there was an error in serialization or not, etc. Using the partition count and sent to partitions information from broadcast return value and the callback to be invoked after each send completion, the caller can implement a wait group if it chooses to implement a stricter broadcast semantics or wants to wait/block for broadcast to complete before progressing. By default broadcast will provide only best-effort semantics. Broadcast throws an exception if it fails to send the event to all partitions due to a runtime exception.

ryannedolan

some nits but looks good

ryannedolan · 2022-04-27T17:23:57Z

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

@@ -231,7 +233,10 @@ public void helperSendOrBroadcast(DatastreamProducerRecord record, SendCallback
          _dynamicMetricsManager.createOrUpdateCounter(MODULE, getDatastreamName(),
              DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1);
          _dynamicMetricsManager.createOrUpdateCounter(MODULE, AGGREGATE, DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1);
-          return;
+          if (isBroadcast) {
+            broadcastMetadata = new DatastreamRecordMetadata(true);


Rather than assign this var to null and then conditionally reassigning, maybe return new DatastreamRecordMetadata(true) here and return null otherwise. Reassigning variables is not great for readability.

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

ryannedolan · 2022-04-27T20:33:40Z

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

    try {
      validateEventRecord(record);
-
      try {


nested try? can probably simplify and have one try-catch pair per method.

It seems a little problematic to refactor this in a neat way.

The try block for serialization error has a catch all Exception and I wan't able to identify the correct set of specific exceptions that might occur, so I introduced a flag to differentiate the exception in the catch block. Other alternative if to have multiple try blocks instead of nested try block. I am personally not partial to either adding a flag or nested try block. But pls let me know if there is a cleaner way that I might have overlooked.

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTestUtils.java

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProvider.java

...api/src/main/java/com/linkedin/datastream/server/api/transport/DatastreamRecordMetadata.java

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

ryannedolan · 2022-05-09T17:56:15Z

...server-api/src/main/java/com/linkedin/datastream/server/api/transport/TransportProvider.java

+   * onEventComplete will be called on completion of record send to each endpoint and each onEventComplete callback will
+   * contain result of send completion to that endpoint.
+   *
+   * If a client wants to build guaranteed broadcast semantics or needs to do additional booking (like which endpoints


s/booking/book keeping/

ryannedolan · 2022-05-09T18:03:24Z

...stream-server-api/src/main/java/com/linkedin/datastream/server/DatastreamProducerRecord.java

+   * Set partition for the record. Can be used to set partition for broadcast record each time before sending.
+   * @param partition Topic partition to send the record to.
+   */
+  public void setPartition(int partition) {


I think we need to keep this class immutable. Are you just trying to avoid a copy? I think that's not worth the broken semantics.

ryannedolan · 2022-05-09T18:20:28Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProvider.java

+    List<Integer> sentToPartitions = new ArrayList<>();
+    try {
+      for (; partition < partitionCount; partition++) {
+        record.setPartition(partition);


I think modifying the record like this will get us into trouble. The caller would not expect the object to be modified, and the internal send() is probably expecting an immutable object. For example, what if send() were to buffer the record before looking at the partition?

I think we need to construct a new record for each send, either by shallow-copying into a new record (which is dangerous, cuz the fields to copy may change over time), or by wrapping the record in an outer class that overrides getPartition. Then you could do:

send(new DatastreamProducerRecordWithPartititionOverride(record, newPartition))

or smth.

Went the copy route

Instead of modifying partition on the DatastreamProducerRecord creating a copy of it for each partition

ryannedolan

thanks, lgtm!

vmaheshw · 2022-05-19T17:26:45Z

...stream-server-api/src/main/java/com/linkedin/datastream/server/DatastreamProducerRecord.java

@@ -20,11 +20,12 @@
 * List of brooklin events that need to be sent to the {@link com.linkedin.datastream.server.api.transport.TransportProvider}
 */
 public class DatastreamProducerRecord {
-  private final Optional<Integer> _partition;
+  private Optional<Integer> _partition;


This change is not required.

surajkn requested review from jzakaryan, ryannedolan, shrinandthakkar, srinagaraj and vmaheshw April 17, 2022 22:51

Add broadcast API to EventProducer

2df1cbe

vmaheshw changed the title ~~Introduce braodcast API to TrasnportProvider~~ Introduce broadcast API to TransportProvider Apr 18, 2022

vmaheshw reviewed Apr 18, 2022

View reviewed changes

vmaheshw reviewed Apr 19, 2022

View reviewed changes

surajkn requested a review from atoomula April 19, 2022 18:02

ryannedolan requested changes Apr 19, 2022

View reviewed changes

ryannedolan previously approved these changes Apr 27, 2022

View reviewed changes

Addressing nits in EventProducer

729b934

surajkn dismissed ryannedolan’s stale review via 729b934 April 28, 2022 17:59

vmaheshw reviewed Apr 28, 2022

View reviewed changes

Addressing nit comments and some refactoring

496d7d6

ryannedolan requested changes May 9, 2022

View reviewed changes

Refactoring to keep DatastreamProducerRecord immutable

8372039

Instead of modifying partition on the DatastreamProducerRecord creating a copy of it for each partition

ryannedolan previously approved these changes May 13, 2022

View reviewed changes

vmaheshw previously approved these changes May 19, 2022

View reviewed changes

Addressing nit. Declaring partition in DatastreamProducerRecord as final

3272aab

surajkn dismissed stale reviews from vmaheshw and ryannedolan via 3272aab May 20, 2022 19:11

ryannedolan approved these changes May 23, 2022

View reviewed changes

vmaheshw approved these changes May 23, 2022

View reviewed changes

surajkn merged commit 55027bc into linkedin:master May 23, 2022

surajkn deleted the tp_broadcast branch May 23, 2022 20:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce broadcast API to TransportProvider #903

Introduce broadcast API to TransportProvider #903

surajkn commented Apr 17, 2022 •

edited by vmaheshw

Loading

ryannedolan left a comment

ryannedolan left a comment

ryannedolan Apr 27, 2022

ryannedolan Apr 27, 2022

surajkn Apr 28, 2022

ryannedolan May 9, 2022

ryannedolan May 9, 2022

ryannedolan May 9, 2022

surajkn May 10, 2022

ryannedolan left a comment

vmaheshw May 19, 2022

Introduce broadcast API to TransportProvider #903

Introduce broadcast API to TransportProvider #903

Conversation

surajkn commented Apr 17, 2022 • edited by vmaheshw Loading

ryannedolan left a comment

Choose a reason for hiding this comment

ryannedolan left a 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

ryannedolan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

surajkn commented Apr 17, 2022 •

edited by vmaheshw

Loading