SAMZA-1856: eventhub descriptors #696

dxichen · 2018-10-06T01:13:30Z

@prateekm @lhaiesp
Add system descriptors and input/output descriptors for Eventhubs.
We should make an effort to deprecate the systems.%s.stream.list.

prateekm · 2018-10-08T16:47:22Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubInputDescriptor.java

+   * @param namespace namespace of the Eventhub entity to consume from
+   * @return this input descriptor
+   */
+  public EventHubInputDescriptor<StreamMessageType> withNamespace(String namespace) {


What does "Required to access the Eventhubs entity per stream" mean? Is this a required configuration to consume from an Event Hubs stream? If so, would be better to make this a param in EventHubsSystemDescriptor#getInputStream(streamId, ...).

The Namespace and Entity for Event Hubs are similar to topics in Kafka. They are required since they indicate the data destination for the output and source for input. I will add those fields in both input and output descriptors.

prateekm · 2018-10-08T16:48:08Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubInputDescriptor.java

+  }
+
+  /**
+   * Namespace associated with the input stream. Required to access the input Eventhubs entity per stream.


Looks like the official name is "Event Hubs"? s/Eventhubs/Event Hubs/g in javadocs and configuration reference.

prateekm · 2018-10-08T16:48:39Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubInputDescriptor.java

+   * @return this input descriptor
+   */
+  public EventHubInputDescriptor<StreamMessageType> withEntityPath(String path) {
+    this.entityPath = Optional.of(path);


Optional.ofNullable(StringUtils.stripToNull(str)) for string params everywhere.

prateekm · 2018-10-08T16:49:44Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubOutputDescriptor.java

+import org.apache.samza.serializers.Serde;
+
+/**
+ * A descriptor for an EventHubs input stream


s/input/output.

prateekm · 2018-10-08T16:50:42Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubOutputDescriptor.java

+   * @param sasKeyName the name of the SAS key required to access the Eventhub entity
+   * @return this output descriptor
+   */
+  public EventHubOutputDescriptor<StreamMessageType> withSasKeyName(String sasKeyName) {


Don't know what SAS key/token is. If this is a kind of auth token, is it possible to configure it per system instead of per stream?

This token could be configured at both a stream level and a system level. Currently we do not support the system level configuration. I have created a ticket (SAMZA-1939) for that config and will update the system descriptors when that is added.

prateekm · 2018-10-08T16:56:22Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

+   * @return this system descriptor
+   */
+  public EventHubSystemDescriptor withPartitioningMethod(PartitioningMethod partitioningMethod) {
+    this.producerEventhubsPartitioningMethod = Optional.of(partitioningMethod);


ofNullable or throw exception if null

prateekm · 2018-10-08T17:00:14Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

+  }
+
+  /**
+   *  Sending each message key to the eventhub in the properties of the AMQP message.


Don't understand the first sentence, seems like an incomplete phrase. Can you clarify, both here and in config reference?

Event Hubs uses eventData, which is a AMQP map under to hood. The kv-pair format from Samza is therefore not kept. Users may specify that they want to add the key in the map in addition to the value. I will clarify

prateekm · 2018-10-08T17:00:29Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

+   * @param sendKeys Set to true if the message key should be send in the EventData properties, false otherwise
+   * @return this system descriptor
+   */
+  public EventHubSystemDescriptor withSendKeys(Boolean sendKeys) {


use boolean so that this can't be null.

prateekm · 2018-10-08T17:00:58Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

+   *   ie. if there are 6 partitions and the key is 9, the message will end up in partition 3.
+   *   Similarly to EVENT_HUB_HASHING, if the partition key is not set the message key is used instead.</li>
+   * </ul>
+   * @param partitioningMethod


Add documentation for @param and @returns everywhere.

prateekm · 2018-10-08T17:01:24Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

+   *  Sending each message key to the eventhub in the properties of the AMQP message.
+   *  If the Samza Eventhub consumer is used, this field is used as the message key if the partition key is not present.
+   *
+   * @param sendKeys Set to true if the message key should be send in the EventData properties, false otherwise


Nitpick: "set to true" (lower case s)

lhaiesp · 2018-10-08T17:06:39Z

Thanks for the work. Agree that we should deprecate the stream.list config. But once we make that change, we need to make sure Brooklin connector adjust to that change as well. Also Brooklin EventHubs connector should use these descriptors.

prateekm

Thanks for cleaning this up. A few more suggestions.

prateekm · 2018-10-08T23:54:25Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubInputDescriptor.java

 import org.apache.samza.serializers.Serde;


 /**
- * A descriptor for the EventHubs output stream
+ * A descriptor for the Event Hubss output stream


Minor: Remove extra s in Event Hubss

prateekm · 2018-10-08T23:55:15Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubInputDescriptor.java

-  public EventHubInputDescriptor(String streamId, Serde serde, SystemDescriptor systemDescriptor,
-      InputTransformer transformer) {
-    super(streamId, serde, systemDescriptor, transformer);
+  EventHubInputDescriptor(String streamId, Serde serde, SystemDescriptor systemDescriptor, String namespace,


Minor: move these two param after streamId and before serde. Same for docs.

prateekm · 2018-10-08T23:57:58Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubOutputDescriptor.java

@@ -48,52 +50,39 @@
   * @param streamId id of the stream
   * @param serde serde for messages in the stream
   * @param systemDescriptor system descriptor this stream descriptor was obtained from
+   * @param namespace namespace for the Event Hubs entity to produce to, not null


Same comment as above, reorder params for logical grouping.

This is resolved, not sure why it still doesn't show the diff in the preview.

prateekm · 2018-10-08T23:58:46Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

  public <StreamMessageType> EventHubInputDescriptor<StreamMessageType> getInputDescriptor(String streamId,
-      Serde<StreamMessageType> serde) {
-    return new EventHubInputDescriptor<>(streamId, serde, this, null);
+      Serde<StreamMessageType> serde, String namespace, String entityPath) {


Reorder params. Since this is not the regular overridden method anymore, also add javadoc for this method including its params. Same below.

prateekm

Final comments.

prateekm · 2018-10-09T18:47:43Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

+ * <p>
+ * System properties configured using a descriptor override corresponding properties provided in configuration.
+ */
+public class EventHubSystemDescriptor extends SystemDescriptor<EventHubSystemDescriptor> {


Minor: Use EventHubsSystemDescriptor (use plural hubs) etc. in all these class names.

prateekm · 2018-10-09T19:04:31Z

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubSystemDescriptor.java

+    super(systemName, FACTORY_CLASS_NAME, null, null);
+  }
+
+  public <StreamMessageType> EventHubInputDescriptor<StreamMessageType> getInputDescriptor(String streamId, String namespace,


Need to add javadoc for public methods, since they don't override the base class methods anymore.

…entation

prateekm

Looks good, thanks!

prateekm requested changes Oct 8, 2018

View reviewed changes

lhaiesp approved these changes Oct 8, 2018

View reviewed changes

prateekm reviewed Oct 9, 2018

View reviewed changes

dxichen force-pushed the eventhub-system-descriptors branch from 887be94 to 4275487 Compare October 9, 2018 00:52

prateekm requested changes Oct 9, 2018

View reviewed changes

dxichen added 6 commits October 9, 2018 15:19

SAMZA-1856 eventhub descriptors

94ca823

Fixed checkstyle

9781c7c

Changed input/output descr params, stream list creation method, docum…

e8fbda2

…entation

reorder params

13bd7df

Eventhub -> Event Hubs in docs

2aa7b09

renamed files, added javadoc

ac244e8

dxichen force-pushed the eventhub-system-descriptors branch from c3e7e98 to ac244e8 Compare October 9, 2018 22:36

deleted duplicates

335dc1f

prateekm approved these changes Oct 11, 2018

View reviewed changes

asfgit closed this in da88096 Oct 11, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SAMZA-1856: eventhub descriptors #696

SAMZA-1856: eventhub descriptors #696

dxichen commented Oct 6, 2018

prateekm Oct 8, 2018

dxichen Oct 8, 2018

prateekm Oct 8, 2018

prateekm Oct 8, 2018

prateekm Oct 8, 2018

prateekm Oct 8, 2018

dxichen Oct 8, 2018

prateekm Oct 8, 2018

prateekm Oct 8, 2018

dxichen Oct 8, 2018

prateekm Oct 8, 2018

prateekm Oct 8, 2018

prateekm Oct 8, 2018

lhaiesp commented Oct 8, 2018

prateekm left a comment

prateekm Oct 8, 2018

prateekm Oct 8, 2018

prateekm Oct 8, 2018

dxichen Oct 9, 2018

prateekm Oct 8, 2018

prateekm left a comment

prateekm Oct 9, 2018

prateekm Oct 9, 2018

prateekm left a comment

SAMZA-1856: eventhub descriptors #696

SAMZA-1856: eventhub descriptors #696

Conversation

dxichen commented Oct 6, 2018

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

lhaiesp commented Oct 8, 2018

prateekm 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

prateekm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

prateekm left a comment

Choose a reason for hiding this comment