[STORM-3045] Microsoft Azure EventHubs Storm Spout and Bolt improvements by SreeramGarlapati · Pull Request #2588 · apache/storm

SreeramGarlapati · 2018-03-07T06:17:53Z

This is continuation of work done by @raviperi.

-update to the latest version of eventhubs java client
-Introduce config params to use latest EH client, control request prefetch size, batch size of events received per call.
-Refactor the code to group classes more appropriately
-Remove redundant types
-Javadoc comments where applicable
-Preftch config parameter to dictate EH prefetch count
-config parameter to introduce sleep between spout's nexttuple calls
-config parameter to retrieve a batched number of events per call to EH
(opposed to single event)
-New data scheme to group event payload and audit params into a single
type, and expose the single type as the only tuple field to downstream
bolts.

Close #2322

of events received per call. -Refactor the code to group classes more appropriately -Remove redundant types -Javadoc comments where applicable -Preftch config parameter to dictate EH prefetch count -config parameter to introduce sleep between spout's nexttuple calls -config parameter to retrieve a batched number of events per call to EH (opposed to single event) -New data scheme to group event payload and audit params into a single type, and expose the single type as the only tuple field to downstream bolts.

… when returning empty/null values

…Config

…upported schemes. Enable scheme based serialization in spout

# Conflicts: # external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBolt.java # external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/core/FieldConstants.java # external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/format/StringEventDataScheme.java # external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventDataScheme.java # external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubException.java # external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java # external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubSpout.java # external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/SpoutOutputCollectorMock.java # pom.xml

…arameter

srdo

Partial review, got down to the start of the Trident code. Will finish up later.

In addition to the other comments, could you please look at target/checkstyle-violations.xml when you've built, and try to resolve any violations that point to your changes? I noticed a number of places with weird indentation, tabs and ifs with no braces, that I'd like to see fixed, since we're trying to stick to the checkstyle rules for new code.

srdo · 2018-03-10T18:54:27Z

+     * @param password      Password to use
+     * @param namespace     target namespace for the service bus
+     * @param entityPath    Name of the event hub
+     * @param partitionMode number of partitions


This description might need to be updated.

srdo · 2018-03-10T18:58:01Z

+    public void prepare(Map<String, Object> config, TopologyContext context,
+                        OutputCollector collector) {
+        this.collector = collector;
+        logger.info(String.format("Conn String: %s, PartitionMode: %s", boltConfig.getConnectionString(),


The logger can do String.format inherently. use {} for placeholders (e.g. logger.info("Connection String: {}", boltConfig.getConnectionString())

srdo · 2018-03-10T19:01:56Z

+ * <p>
+ * The implementation has two modes of operation:
+ * <ul>
+ * <li>partitionmode = true, One bolt for per partition write.</li>


Nit: This seems like an enum would be a good fit.

srdo · 2018-03-10T19:08:13Z

+            ehClient.close().whenCompleteAsync((voidargs, error) -> {
+                try {
+                    if (error != null) {
+                        logger.error("Exception during EventHubBolt cleanup phase" + error.toString());


Consider logging the exception without toString so you don't lose the stack trace, i.e. logger.error("Some message", error). This looks like it's a problem in a few places.

srdo · 2018-03-10T19:10:43Z

+    public void cleanup() {
+        logger.debug("EventHubBolt cleanup");
+        try {
+            ehClient.close().whenCompleteAsync((voidargs, error) -> {


Nit: Any reason to use whenCompleteAsync over whenComplete?

srdo · 2018-03-10T22:03:16Z

+    @Override
+    public void declareOutputFields(final OutputFieldsDeclarer declarer) {
+        List<String> fields = new LinkedList<String>();
+        fields.add(FieldConstants.MESSAGE_FIELD);


You should use the fields declared by the scheme here, otherwise what's the point of having IEventDataScheme provide a declareOutputFields?

srdo · 2018-03-10T22:10:11Z

+import org.apache.storm.eventhubs.core.EventHubConfig;
+
+/**
+ * EventHub configuration. This class remains in


Nit: Didn't the other class moves already break binary compatibility?

srdo · 2018-03-10T22:11:57Z

+    private static final String ZK_LOCAL_URL = "localhost:2181";
+
+    private final String zookeeperConnectionString;
+    private final CuratorFramework curatorFramework;


I don't think this is serializable

srdo · 2018-03-10T22:15:38Z

+                return null;
+            }
+
+            String data = new String(curatorFramework.getData().forPath(statePath));


Better if you specify a charset

srdo · 2018-03-10T22:15:54Z

+    @Override
+    public void saveData(String statePath, String data) {
+        data = StringUtil.isNullOrWhiteSpace(data) ? "" : data;
+        byte[] bytes = data.getBytes();


You should specify a charset

SreeramGarlapati · 2018-03-11T00:12:27Z

thanks @srdo - i will take a shot at your suggestions by EOD monday PST.

SreeramGarlapati · 2018-03-23T23:37:45Z

@srdo - when I try to merge my changes with master - I ended up with a merge conflicts due to line endings change. is there any simple tip - which I can use to ease through this? like changed my files to a specific line ending before the merge etc?
Thanks a lot for the help so far!

srdo · 2018-03-24T08:36:12Z

@SreeramGarlapati Quick google suggests that you can try adding -Xrenormalize to the merge command. https://stackoverflow.com/a/12194759/8845188

srdo · 2018-03-24T08:36:48Z

And also, yes you can change the line ending to LF and that should work too

srdo · 2018-03-24T12:48:45Z

By the way, when you get a chance, please go to https://issues.apache.org/jira and create an issue for tracking these changes, then rename the PR and commit(s) so they contain the issue number. It makes it easier for us to track which branches the changes are applied to, and helps us generate correct release notes. Thanks.

…ation

raviperi and others added 15 commits September 11, 2017 14:35

Remove unncessary file.

3f8f37f

Add license header, remove next tuple sleep time as it is redundant.

9a18f9a

Remove unused nextTuple sleep interval field

5a7eb2e

Add javadoc for EventHubConfig. Remove unused fields. Improve logging…

6ba9d11

… when returning empty/null values

Organize usings. Remove extra constructor and fields in EventHubSpout…

53569ac

…Config

Remove obselete Binary and EventDataSchemes, and add unit tests for s…

b1bde9b

…upported schemes. Enable scheme based serialization in spout

Mark all fields final in EventHUbMessage

4516d01

Update batch size log message to debug level

2063f45

remove files as continuation to the merge commit

a61466d

move the library to 1.0.0

19bd1c9

CheckStyle: refactor code

dffb8a5

fix unit tests

32ea9d1

fix an issue in eventhubbolt constructor - to not ignore entityPath p…

b69fcf3

…arameter

srdo reviewed Mar 10, 2018

View reviewed changes

SreeramGarlapati added 2 commits March 23, 2018 13:45

open EventHubSpout for inheritance

d468939

update eventhubs client to latest

5974bfb

address cr feedback - partial - checkpoint progress

6ee2860

SreeramGarlapati changed the title ~~Microsoft Azure EventHubs Storm Spout and Bolt improvements~~ [STORM-3045] Microsoft Azure EventHubs Storm Spout and Bolt improvements Apr 27, 2018

SreeramGarlapati added 5 commits May 2, 2018 16:44

Mark EventHubMessage Serializable

04603b2

Mark MessageId as Serializable

d3d9fe2

Fix ambiguity/bug in PartitionManager.Checkpoint implementation

5c94db6

Implement PartitionManager honoring STORM's pendingQueueSize configur…

1f3195c

…ation

fix issue with checkpointing.

bbc5f0c

SreeramGarlapati and others added 7 commits May 8, 2018 11:49

remove strictorderpartitionmanager

a5a0efb

Fix serialization issues in EventHubSpout class

087c409

minor refactor

de3814f

adjust prefetch relevant to MaxPendingSize

6b528ca

tracing improvements

8d0dfa1

nit refactor

927360a

minor refactor

577d4ee

SreeramGarlapati closed this May 21, 2021

Uh oh!

Conversation

SreeramGarlapati commented Mar 7, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

srdo left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SreeramGarlapati commented Mar 11, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SreeramGarlapati commented Mar 23, 2018

Uh oh!

srdo commented Mar 24, 2018

Uh oh!

srdo commented Mar 24, 2018

Uh oh!

srdo commented Mar 24, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

SreeramGarlapati commented Mar 7, 2018 •

edited

Loading

SreeramGarlapati commented Mar 11, 2018 •

edited

Loading