This patch is still under development and was uploaded at this moment for early testing. Please read README.

There may be a bug in the offsets management, because of diff o 1. I am looking into it. Currently polling from an arbitrary offset is possible but it will come in the next patch, today or tomorrow I refactored the code a bit and left, maybe, some unnecessary locking. I am also looking into it.

@connieyang @jianbzhou @tgravescs please let me know of any other requirements you may have and I will address them soon.

Thanks

Description can be found at https://issues.apache.org/jira/browse/STORM-329 The main change are:

1. set a client buffer limit, if the buffer message exceed the limit, it will wait util there are enough space for the new arrival message.
2. if we lose the connection with remote worker, the reconnect logic will be asynchronously and will not block other worker.
3. there are timer for cache messages, if the messages are cache for more than "topology.message.timeout.secs", we will drop the cache message.

For large cluster support, good performance for scheduling and assignments distribution.

For nimbus heartbeats pressure test Heartbeats response time: The horizontal ordinate is number of nodes/supervisors which report heartbeats every 1 second, for our production cluster, heartbeat reporting interval is 5 seconds and time out 30 seconds, so this means new heartbeats mode supports at least 5 * 2000 nodes.

For topology submission time cost Topology submission scheduling time:

cluster used slots | newly submit workers | after submission used slots | time cost milliseconds -- | -- | -- | -- 3700 | 100 | 3800 | 1181ms 3600 | 100 | 3700 | 886ms 3500 | 100 | 3600 | 925ms 3400 | 100 | 3500 | 1520ms 3300 | 100 | 3400 | 930ms 2000 | 100 | 2100 |  722ms 1500 | 100 | 1600 |  747ms 1000 | 100 | 1100 |  754ms 500 | 100 | 600 |  695ms 100 | 100 | 200 |  749ms

This is PR for 1.1.x-branch of heartbeats and assignments promotion #2389

This is PR for 1.1.x-branch of assignments promotion [closed now] #2319

Having spent a lot of time on this, I am happy to share some good news and some even better news with you.

Before venturing further, I must add, to limit the scope of this PR, no attempt was made to improve ACK-ing mode perf. Although there are some big latency gains seen in ACK mode, these are a side effect of the new messaging design and work remains to be done to improve ACK mode.

Please see the design docs posted on the STORM-2306 jira for details on what is being done

So, first the good news .. a quick competitive evaluation:

1) Competitive Perf evaluation :

Here are some quick comparison of Storm numbers taken on my laptop against numbers for similar/identical topologies published by Heron, Flink and Apex. Shall provide just rolled up summary here and leave the detailed analysis for later.

Storm numbers here were run on my MacBook Pro (2015) with 16GB ram and a single 4 core Intel i7 chip.

A) Compared To Heron and Flink:

Heron recently published this blog about the big perf improvements (~4-6x) they achieved. https://blog.twitter.com/engineering/en_us/topics/open-source/2017/optimizing-twitter-heron.html They ran it on dual 12-core Intel Xeon chips (didn't say how many machines).

They use a simplified word count topology that I have emulated for comparison purposes and included it as part of this PR (SimplifiedWordCountTopo).

Flink also publishes numbers for a similar setup here https://flink.apache.org/features.html#streaming

Below are per core throughput numbers.

[:HERON:] Acking Disabled: per core = ~475 k/sec. Acking Enabled: per core = ~150 k/sec. Latency = 30ms

[:FLINK:] Per core: ~1 mill/sec

[:STORM:] Acking Disabled: per core = 2 mill/sec. (1 spout & 1 counter bolt) Acking Enabled: per core = 0.6 mill/sec, Latency = 0.73 ms (+1 acker)

Takeaways:

• Storm's with-ACK throughput is better than Heron's no-ACK throughput.
• Without ACKs, Storm is far ahead of Heron and also better than Flink.
• Storm's Latency (in microseconds) is also significantly better than both (although this metric is better to compared with multiple machines in the run). AFAIKT, Flink is generally not known for having good latency (as per Flink's own comparison with Storm - https://data-artisans.com/blog/high-throughput-low-latency-and-exactly-once-stream-processing-with-apache-flink).

B) Compared to Apex:

Apex appears to be the best performer among the opensource lot.. by a reasonably good margin. Some numbers they published in their early days (when it was called DataTorrent) were misleading/dubious IMO, but the newer numbers appear credible.

Here we look at how fast inter spout/bolt communication can be achieved using an ultra minimalist topology. A ConstSpout emits a short string to a DevNull bolt that discards the tuples it receives. This topo has been in storm-perf for sometime now.

Apex provides numbers for a identical setup ... what they call "container local" performance here: https://www.datatorrent.com/blog/blog-apex-performance-benchmark/

Other than the fact that Storm numbers were on my laptop, these numbers are a good apples to apples comparison.

[:APEX:] Container local Throughput : ~4.2 mill/sec

[:STORM:] Worker local throughput : 8.1 mill/sec

2) Core messaging Performance

Now for the better news. The redesigned messaging system is actually much faster and able to move messages between threads at an astounding rate .... :

• 120 mill/sec (batchSz=1000, 2 producers writing to 1 consumer).
• 67 mill/sec (batchSz=1000, 1 producers writing to 1 consumer).

I have included JCQueuePerfTest.java in this PR to help get quick measurements from within the IDE.

That naturally begs the question .. why is Storm pushing only 8.1 mill/sec between a ConstSpout and DevNullBolt ? The short answer is ... there are big bottlenecks in other parts of the code. In this PR I have tackled some such bottlenecks but many still remain. We are faster than the competition, but still have room to be much much faster. If anyone is interested in pursuing these to push Storm's perf to the next level, I am happy to point them in the right direction.

Again, please refer to the design docs in the JIRA for details on the new design and the rationale behind them.

STORM-297

Description and test report can be found at https://issues.apache.org/jira/browse/STORM-297 The changes consists of:

1. use netty async
2. use batch send and batch receiver messaging api
3. allow to configure multiple worker receiver threads.
4. name the executor and netty threads

1. removed html rendering from core.clj
2. added RESTful apis to core.clj
3. moved html rendering to jquery & mustache on top of newly added restful apis
4. added api
   1. /api/cluster/configuration
   2. /api/cluster/summary
   3. /api/supervisor/summary
   4. /api/topology/summary
   5. /api/topology/:id
   6. /api/topology/:id/component/:component
   7. /api/topology/:id/activate (POST)
   8. /api/topology/:id/deactivate (POST)
   9. /api/topology/:id/rebalance/:wait-time (POST)
   10. /api/topology/:id/kill/:wait-time (POST) all of the above methods returns json response.

Still needs run as user and CGroup work, but the rest is working

Any feedback on this would be welcome. I am particularly interested in the readability of the code and how easy is it to understand vs the original supervisor.

All the unit tests pass and I have done a lot of manual testing switching back and forth between the original supervisor and supervisor V2 to ensure that this can be a rolling upgrade.

There are still a number of TODOs in the code. Most of them are for removing the original supervisor code and cleaning up the result of that.

This PR contains the same code as https://github.com/apache/storm/pull/428 but as a single commit for a cleaner commit history of our Storm repo.

This is an improved version of the original pull request discussed at https://github.com/apache/storm/pull/268. Please refer to the discussion in the link above.

The changes of this pull request include:

• Most importantly, we fix a bug in Storm that may cause a cascading failure in a Storm cluster, to the point where the whole cluster becomes unusable. This is achieved by the work described in the next bullet points.
• We refactor and improve the Netty messaging backend, notably the client.
• During the initial startup of a topology, Storm will now wait until worker (Netty) connections are ready for operation. See the original discussion thread for the detailed explanation and justification of this change.

@clockfly, @tedxia: Please add any further comments to STORM-329 to this pull request, if possible.

I have deleted the bit torrent implementation from this pull request as the only available bit torrent library does not support tracker less torrents. In absence of tracker less torrents a single tracker becomes a single point of failure and a multi tracker implementation requires that if a tracker host fails the replacement host has same dns/network configuration.

Some manual tests I executed:

• start 3 nimbuses, test simple word count topology works. try storm list/activate/deactivate/rebalance/kill from ui and CLI.
• set the replication factor to 2 run the first test again.
• bring up a new nimbus, ensure it catches up and competes for leader lock.
• with 3 nimbuses and 2 topologies, delete one topology code from each non leader nimbus. After killing master nimbus, ensure one of them eventually becomes leader.

• JIRA issue: http://issues.apache.org/jira/browse/STORM-2016
• design doc: https://cwiki.apache.org/confluence/display/STORM/A.+Design+doc%3A+adding+jars+and+maven+artifacts+at+submission
• discussion thread: http://mail-archives.apache.org/mod_mbox/storm-dev/201608.mbox/%3CCAF5108i9+tJaNZ0LgRkTMkVQEL7F+53k9uyzxcT6zhSU6OHx9Q@mail.gmail.com%3E

• bin/storm now supports "--jars" and "--packages" options
  • it's only effective with "storm jar" and "storm sql"
• introduce new module: storm-submit to help resolving dependencies with handling transitive dependencies
• StormSubmitter will upload dependencies to BlobStore when submitting topology
• Supervisor will download dependencies from BlobStore when such topology is assigned
• Supervisor will launch workers with adding downloaded dependencies to worker classpath

TODO

• [x] documentation
• [x] craft pull request against master branch

Btw, it might be better to place some modules out of external since 'external' has most of non storm-core modules and flux, sql, storm-submit, storm-kafka-monitor are not kind of connectors.

The throughputs for the spouts and bolts could help the user to identify the performance bottleneck and detect the load balancing issue. In this RP, I take measurements on the throughput of the executors and display them on web UI.

Summary of Changes

1. Take throughput measurements on the spouts and bolts;
2. Add throughput to ExecutorStats;
3. Display the throughputs on web UI.

Note: If you cannot see the throughputs on your web UI, please clean your browser cache and try again.

Screenshots

   public OutputStream getOutputStream() throws IOException {
        FsPermission fileperms = new FsPermission(BLOBSTORE_FILE_PERMISSION);
        try {
            out = fileSystem.create(path, (short) this.getMetadata().get_replication_factor());
            fileSystem.setPermission(path, fileperms);
            fileSystem.setReplication(path, (short) this.getMetadata().get_replication_factor());
        } catch (IOException e) {
           ......
            out = fileSystem.create(path, (short) this.getMetadata().get_replication_factor());
            fileSystem.setPermission(path, dirperms);
            fileSystem.setReplication(path, (short) this.getMetadata().get_replication_factor());
        }
        ......
    }

We can see that there are permission settings for path in both try and catch, but the permission in catch is different from that in try. In catch, the permission dirperms is given to the file. I think there is a problem here, and it should be the same as The permissions in try are consistent. Permissions should be set according to the following code

 public OutputStream getOutputStream() throws IOException {
        FsPermission fileperms = new FsPermission(BLOBSTORE_FILE_PERMISSION);
        try {
            out = fileSystem.create(path, (short) this.getMetadata().get_replication_factor());
            fileSystem.setPermission(path, fileperms);
            fileSystem.setReplication(path, (short) this.getMetadata().get_replication_factor());
        } catch (IOException e) {
           ......
            out = fileSystem.create(path, (short) this.getMetadata().get_replication_factor());
            fileSystem.setPermission(path, fileperms);
            fileSystem.setReplication(path, (short) this.getMetadata().get_replication_factor());
        }
        ......
    }

What happened？

There are 2 security vulnerabilities found in com.fasterxml.jackson.core:jackson-databind 2.10.5.1

• CVE-2020-36518
• MPS-2022-12500

What did I do？

Upgrade com.fasterxml.jackson.core:jackson-databind from 2.10.5.1 to 2.12.6.1 for vulnerability fix

What did you expect to happen？

Ideally, no insecure libs should be used.

The specification of the pull request

PR Specification from OSCS

What happened？

There are 2 security vulnerabilities found in com.google.guava:guava 17.0

• CVE-2018-10237
• CVE-2020-8908

What did I do？

Upgrade com.google.guava:guava from 17.0 to 30.0-jre for vulnerability fix

What did you expect to happen？

Ideally, no insecure libs should be used.

The specification of the pull request

PR Specification from OSCS

What happened？

There are 6 security vulnerabilities found in org.elasticsearch:elasticsearch 2.4.4

• CVE-2020-7020
• CVE-2021-22144
• CVE-2020-7021
• CVE-2021-22137
• CVE-2021-22135
• CVE-2019-7614

What did I do？

Upgrade org.elasticsearch:elasticsearch from 2.4.4 to 6.8.17 for vulnerability fix

What did you expect to happen？

Ideally, no insecure libs should be used.

The specification of the pull request

PR Specification from OSCS

What happened？

There are 1 security vulnerabilities found in org.fusesource.mqtt-client:mqtt-client 1.10

• CVE-2019-0222

What did I do？

Upgrade org.fusesource.mqtt-client:mqtt-client from 1.10 to 1.15 for vulnerability fix

What did you expect to happen？

Ideally, no insecure libs should be used.

The specification of the pull request

PR Specification from OSCS