| <!DOCTYPE html><html lang="en"><head><meta charSet="utf-8"/><meta http-equiv="X-UA-Compatible" content="IE=edge"/><title>The Heron Topology API for Java · Apache Heron</title><meta name="viewport" content="width=device-width"/><meta name="generator" content="Docusaurus"/><meta name="description" content="<!--"/><meta name="docsearch:version" content="next"/><meta name="docsearch:language" content="en"/><meta property="og:title" content="The Heron Topology API for Java · Apache Heron"/><meta property="og:type" content="website"/><meta property="og:url" content="https://heron.apache.org/"/><meta property="og:description" content="<!--"/><meta property="og:image" content="https://heron.apache.org/img/undraw_online.svg"/><meta name="twitter:card" content="summary"/><meta name="twitter:image" content="https://heron.apache.org/img/undraw_tweetstorm.svg"/><link rel="shortcut icon" href="/img/favicon-32x32.png"/><link rel="stylesheet" href="//cdnjs.cloudflare.com/ajax/libs/highlight.js/9.12.0/styles/default.min.css"/><link rel="alternate" type="application/atom+xml" href="https://heron.apache.org/blog/atom.xml" title="Apache Heron Blog ATOM Feed"/><link rel="alternate" type="application/rss+xml" href="https://heron.apache.org/blog/feed.xml" title="Apache Heron Blog RSS Feed"/><script> |
| (function(i,s,o,g,r,a,m){i['GoogleAnalyticsObject']=r;i[r]=i[r]||function(){ |
| (i[r].q=i[r].q||[]).push(arguments)},i[r].l=1*new Date();a=s.createElement(o), |
| m=s.getElementsByTagName(o)[0];a.async=1;a.src=g;m.parentNode.insertBefore(a,m) |
| })(window,document,'script','https://www.google-analytics.com/analytics.js','ga'); |
| |
| ga('create', 'UA-198017384-1', 'auto'); |
| ga('send', 'pageview'); |
| </script><script type="text/javascript" src="https://buttons.github.io/buttons.js"></script><script type="text/javascript" src="/js/custom.js"></script><script type="text/javascript" src="/js/fix-location.js"></script><link rel="stylesheet" href="/css/main.css"/><script src="/js/codetabs.js"></script></head><body class="sideNavVisible separateOnPageNav"><div class="fixedHeaderContainer"><div class="headerWrapper wrapper"><header><a href="/"><img class="logo" src="/img/HeronTextLogo-small.png" alt="Apache Heron"/><h2 class="headerTitleWithLogo">Apache Heron</h2></a><a href="/versions"><h3>next</h3></a><div class="navigationWrapper navigationSlider"><nav class="slidingNav"><ul class="nav-site nav-site-internal"><li class=""><a href="/api/java" target="_self">Javadocs</a></li><li class=""><a href="/api/python" target="_self">Pydocs</a></li><li class="siteNavGroupActive"><a href="/docs/next/getting-started-local-single-node" target="_self">Docs</a></li><li class=""><a href="/download" target="_self">Downloads</a></li><li class=""><a href="#community" target="_self">Community</a></li><li class=""><a href="/blog/" target="_self">Blog</a></li><li class=""><a href="#apache" target="_self">Apache</a></li></ul></nav></div></header></div></div><div class="navPusher"><div class="docMainWrapper wrapper"><div class="container docsNavContainer" id="docsNav"><nav class="toc"><div class="toggleNav"><section class="navWrapper wrapper"><div class="navBreadcrumb wrapper"><div class="navToggle" id="navToggler"><div class="hamburger-menu"><div class="line1"></div><div class="line2"></div><div class="line3"></div></div></div><h2><i>›</i><span>Topology Development APIs</span></h2><div class="tocToggler" id="tocToggler"><i class="icon-toc"></i></div></div><div class="navGroups"><div class="navGroup"><h3 class="navGroupCategoryTitle">Getting Started</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/getting-started-local-single-node">Local (Single Node)</a></li><li class="navListItem"><a class="navItem" href="/docs/next/getting-started-migrate-storm-topologies">Migrate Storm Topologies</a></li><li class="navListItem"><a class="navItem" href="/docs/next/getting-started-docker">Heron & Docker</a></li><li class="navListItem"><a class="navItem" href="/docs/next/getting-started-troubleshooting-guide">Troubleshooting Guide</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Deployment</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/deployment-overview">Deployment Overiew</a></li><li class="navListItem"><a class="navItem" href="/docs/next/deployment-configuration">Configuration</a></li><li class="navListItem"><a class="navItem" href="/docs/next/deployment-api-server">The Heron API Server</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Topology Development APIs</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/topology-development-streamlet-api">The Heron Streamlet API for Java</a></li><li class="navListItem"><a class="navItem" href="/docs/next/topology-development-eco-api">The ECO API for Java</a></li><li class="navListItem navListItemActive"><a class="navItem" href="/docs/next/topology-development-topology-api-java">The Heron Topology API for Java</a></li><li class="navListItem"><a class="navItem" href="/docs/next/topology-development-topology-api-python">The Heron Topology API for Python</a></li><li class="navListItem"><a class="navItem" href="/docs/next/topology-development-streamlet-scala">The Heron Streamlet API for Scala</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Client API Docs</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/client-api-docs-overview">Client API Docs</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Guides</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/guides-effectively-once-java-topologies">Effectively Once Java Topologies</a></li><li class="navListItem"><a class="navItem" href="/docs/next/guides-data-model">Heron Data Model</a></li><li class="navListItem"><a class="navItem" href="/docs/next/guides-tuple-serialization">Tuple Serialization</a></li><li class="navListItem"><a class="navItem" href="/docs/next/guides-ui-guide">Heron UI Guide</a></li><li class="navListItem"><a class="navItem" href="/docs/next/guides-topology-tuning">Topology Tuning Guide</a></li><li class="navListItem"><a class="navItem" href="/docs/next/guides-packing-algorithms">Packing Algorithms</a></li><li class="navListItem"><a class="navItem" href="/docs/next/guides-simulator-mode">Simulator Mode</a></li><li class="navListItem"><a class="navItem" href="/docs/next/guides-troubeshooting-guide">Topology Troubleshooting Guide</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Heron Concepts</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/heron-design-goals">Heron Design Goals</a></li><li class="navListItem"><a class="navItem" href="/docs/next/heron-topology-concepts">Heron Topologies</a></li><li class="navListItem"><a class="navItem" href="/docs/next/heron-streamlet-concepts">Heron Streamlets</a></li><li class="navListItem"><a class="navItem" href="/docs/next/heron-architecture">Heron Architecture</a></li><li class="navListItem"><a class="navItem" href="/docs/next/heron-delivery-semantics">Heron Delivery Semantics</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">State Managers</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/state-managers-zookeeper">Zookeeper</a></li><li class="navListItem"><a class="navItem" href="/docs/next/state-managers-local-fs">Local File System</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Uploaders</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/uploaders-local-fs">Local File System</a></li><li class="navListItem"><a class="navItem" href="/docs/next/uploaders-hdfs">HDFS</a></li><li class="navListItem"><a class="navItem" href="/docs/next/uploaders-http">HTTP</a></li><li class="navListItem"><a class="navItem" href="/docs/next/uploaders-amazon-s3">Amazon S3</a></li><li class="navListItem"><a class="navItem" href="/docs/next/uploaders-scp">Secure Copy (SCP)</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Schedulers</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-k8s-by-hand">Kubernetes by hand</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-k8s-with-helm">Kubernetes with Helm</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-k8s-execution-environment">Kubernetes Environment Customization</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-aurora-cluster">Aurora Cluster</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-aurora-local">Aurora Locally</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-local">Local Cluster</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-nomad">Nomad</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-mesos-local-mac">Mesos Cluster Locally</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-slurm">Slurm Cluster</a></li><li class="navListItem"><a class="navItem" href="/docs/next/schedulers-yarn">YARN Cluster</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Cluster Configuration</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/cluster-config-overview">Cluster Config Overview</a></li><li class="navListItem"><a class="navItem" href="/docs/next/cluster-config-system-level">System Level Configuration</a></li><li class="navListItem"><a class="navItem" href="/docs/next/cluster-config-instance">Heron Instance</a></li><li class="navListItem"><a class="navItem" href="/docs/next/cluster-config-metrics">Metrics Manager</a></li><li class="navListItem"><a class="navItem" href="/docs/next/cluster-config-stream">Stream Manager</a></li><li class="navListItem"><a class="navItem" href="/docs/next/cluster-config-tmanager">Topology Manager</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Observability</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/observability-prometheus">Prometheus</a></li><li class="navListItem"><a class="navItem" href="/docs/next/observability-graphite">Graphite</a></li><li class="navListItem"><a class="navItem" href="/docs/next/observability-scribe">Scribe</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">User Manuals</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/user-manuals-heron-cli">Heron Client</a></li><li class="navListItem"><a class="navItem" href="/docs/next/user-manuals-heron-explorer">Heron Explorer</a></li><li class="navListItem"><a class="navItem" href="/docs/next/user-manuals-tracker-rest">Heron Tracker REST API</a></li><li class="navListItem"><a class="navItem" href="/docs/next/user-manuals-heron-tracker-runbook">Heron Tracker Runbook</a></li><li class="navListItem"><a class="navItem" href="/docs/next/user-manuals-heron-ui-runbook">Heron UI Runbook</a></li><li class="navListItem"><a class="navItem" href="/docs/next/user-manuals-heron-shell">Heron Shell</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Compiling</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/compiling-overview">Compiling Overview</a></li><li class="navListItem"><a class="navItem" href="/docs/next/compiling-linux">Compiling on Linux</a></li><li class="navListItem"><a class="navItem" href="/docs/next/compiling-osx">Compiling on OS X</a></li><li class="navListItem"><a class="navItem" href="/docs/next/compiling-docker">Compiling With Docker</a></li><li class="navListItem"><a class="navItem" href="/docs/next/compiling-running-tests">Running Tests</a></li><li class="navListItem"><a class="navItem" href="/docs/next/compiling-code-organization">Code Organization</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Extending Heron</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/extending-heron-scheduler">Custom Scheduler</a></li><li class="navListItem"><a class="navItem" href="/docs/next/extending-heron-metric-sink">Custom Metrics Sink</a></li></ul></div><div class="navGroup"><h3 class="navGroupCategoryTitle">Heron Resources</h3><ul class=""><li class="navListItem"><a class="navItem" href="/docs/next/heron-resources-resources">Heron Resources</a></li></ul></div></div></section></div><script> |
| var coll = document.getElementsByClassName('collapsible'); |
| var checkActiveCategory = true; |
| for (var i = 0; i < coll.length; i++) { |
| var links = coll[i].nextElementSibling.getElementsByTagName('*'); |
| if (checkActiveCategory){ |
| for (var j = 0; j < links.length; j++) { |
| if (links[j].classList.contains('navListItemActive')){ |
| coll[i].nextElementSibling.classList.toggle('hide'); |
| coll[i].childNodes[1].classList.toggle('rotate'); |
| checkActiveCategory = false; |
| break; |
| } |
| } |
| } |
| |
| coll[i].addEventListener('click', function() { |
| var arrow = this.childNodes[1]; |
| arrow.classList.toggle('rotate'); |
| var content = this.nextElementSibling; |
| content.classList.toggle('hide'); |
| }); |
| } |
| |
| document.addEventListener('DOMContentLoaded', function() { |
| createToggler('#navToggler', '#docsNav', 'docsSliderActive'); |
| createToggler('#tocToggler', 'body', 'tocActive'); |
| |
| var headings = document.querySelector('.toc-headings'); |
| headings && headings.addEventListener('click', function(event) { |
| var el = event.target; |
| while(el !== headings){ |
| if (el.tagName === 'A') { |
| document.body.classList.remove('tocActive'); |
| break; |
| } else{ |
| el = el.parentNode; |
| } |
| } |
| }, false); |
| |
| function createToggler(togglerSelector, targetSelector, className) { |
| var toggler = document.querySelector(togglerSelector); |
| var target = document.querySelector(targetSelector); |
| |
| if (!toggler) { |
| return; |
| } |
| |
| toggler.onclick = function(event) { |
| event.preventDefault(); |
| |
| target.classList.toggle(className); |
| }; |
| } |
| }); |
| </script></nav></div><div class="container mainContainer"><div class="wrapper"><div class="post"><header class="postHeader"><h1 class="postHeaderTitle">The Heron Topology API for Java</h1></header><article><div><span><!-- |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you under the Apache License, Version 2.0 (the |
| "License"); you may not use this file except in compliance |
| with the License. You may obtain a copy of the License at |
| http://www.apache.org/licenses/LICENSE-2.0 |
| Unless required by applicable law or agreed to in writing, |
| software distributed under the License is distributed on an |
| "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| KIND, either express or implied. See the License for the |
| specific language governing permissions and limitations |
| under the License. |
| --> |
| <p>A topology specifies components like spouts and bolts, as well as the relation |
| between components and proper configurations. The |
| <a href="http://search.maven.org/#search%7Cgav%7C1%7Cg%3A%22org.apache.heron%22%20AND%20a%3A%22heron-api%22"><code>heron-api</code></a> |
| enables you to create topology logic in Java.</p> |
| <blockquote> |
| <p>If you're interested in creating stateful topologies with <a href="heron-delivery-semantics">effectively-once |
| semantics</a> in Java, see <a href="guides-effectively-once-java-topologies">this new |
| guide</a>.</p> |
| </blockquote> |
| <h2><a class="anchor" aria-hidden="true" id="getting-started"></a><a href="#getting-started" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Getting started</h2> |
| <p>In order to use the Heron API for Java, you'll need to install the <code>heron-api</code> library, which is available |
| via <a href="http://search.maven.org/">Maven Central</a>.</p> |
| <h3><a class="anchor" aria-hidden="true" id="maven-setup"></a><a href="#maven-setup" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Maven setup</h3> |
| <p>To install the <code>heron-api</code> library using Maven, add this to the <code>dependencies</code> block of your <code>pom.xml</code> |
| configuration file:</p> |
| <pre><code class="hljs css language-xml"><span class="hljs-tag"><<span class="hljs-name">dependency</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">groupId</span>></span>org.apache.heron<span class="hljs-tag"></<span class="hljs-name">groupId</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">artifactId</span>></span>heron-api<span class="hljs-tag"></<span class="hljs-name">artifactId</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">version</span>></span>0.20.5-incubating<span class="hljs-tag"></<span class="hljs-name">version</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">dependency</span>></span> |
| </code></pre> |
| <h4><a class="anchor" aria-hidden="true" id="compiling-a-jar-with-dependencies"></a><a href="#compiling-a-jar-with-dependencies" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Compiling a JAR with dependencies</h4> |
| <p>In order to run a Java topology in a Heron cluster, you'll need to package your topology as a "fat" JAR with dependencies included. You can use the <a href="https://maven.apache.org/plugins/maven-assembly-plugin/usage.html">Maven Assembly Plugin</a> to generate JARs with dependencies. To install the plugin and add a Maven goal for a single JAR, add this to the <code>plugins</code> block in your <code>pom.xml</code>:</p> |
| <pre><code class="hljs css language-xml"><span class="hljs-tag"><<span class="hljs-name">plugin</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">artifactId</span>></span>maven-assembly-plugin<span class="hljs-tag"></<span class="hljs-name">artifactId</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">configuration</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">descriptorRefs</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">descriptorRef</span>></span>jar-with-dependencies<span class="hljs-tag"></<span class="hljs-name">descriptorRef</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">descriptorRefs</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">archive</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">manifest</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">mainClass</span>></span><span class="hljs-tag"></<span class="hljs-name">mainClass</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">manifest</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">archive</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">configuration</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">executions</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">execution</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">id</span>></span>make-assembly<span class="hljs-tag"></<span class="hljs-name">id</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">phase</span>></span>package<span class="hljs-tag"></<span class="hljs-name">phase</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">goals</span>></span> |
| <span class="hljs-tag"><<span class="hljs-name">goal</span>></span>single<span class="hljs-tag"></<span class="hljs-name">goal</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">goals</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">execution</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">executions</span>></span> |
| <span class="hljs-tag"></<span class="hljs-name">plugin</span>></span> |
| </code></pre> |
| <p>Once your <code>pom.xml</code> is properly set up, you can compile the JAR with dependencies using this command:</p> |
| <pre><code class="hljs css language-bash">$ mvn package |
| </code></pre> |
| <p>By default, this will add a JAR in your project's <code>target</code> folder with the name <code>PROJECT-NAME-VERSION-jar-with-dependencies.jar</code>. Here's an example topology submission command using a compiled JAR:</p> |
| <pre><code class="hljs css language-bash">$ mvn package |
| $ heron submit <span class="hljs-built_in">local</span> \ |
| target/my-project-1.2.3-jar-with-dependencies.jar \ |
| com.example.Main \ |
| MyTopology arg1 arg2 |
| </code></pre> |
| <h3><a class="anchor" aria-hidden="true" id="writing-your-topology-logic"></a><a href="#writing-your-topology-logic" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Writing your topology logic</h3> |
| <p>Heron <a href="heron-topology-concpets">topologies</a> are processing graphs consisting |
| of spouts that ingest data and bolts that process that data.</p> |
| <blockquote> |
| <p><strong>Don't want to manually create spouts and bolts? Try the Heron Streamlet API.</strong> If you find manually creating and connecting spouts and bolts to be overly cumbersome, we recommend trying out the <a href="topology-development-streamlet-api-java">Heron Streamlet API</a> for Java, which enables you to create your topology logic using a highly streamlined logic inspired by functional programming concepts.</p> |
| </blockquote> |
| <p>Once you've defined the spouts and bolts, a topology can be composed using a |
| <a href="/api/org/apache/heron/api/topology/TopologyBuilder.html"><code>TopologyBuilder</code></a>. The |
| <code>TopologyBuilder</code> has two major methods used to specify topology components:</p> |
| <table> |
| <thead> |
| <tr><th style="text-align:left">Method</th><th style="text-align:left">Description</th></tr> |
| </thead> |
| <tbody> |
| <tr><td style="text-align:left"><code>setBolt(String id, IRichBolt bolt, Number parallelismHint)</code></td><td style="text-align:left"><code>id</code> is the unique identifier that assigned to a bolt, <code>bolt</code> is the one previously composed, and <code>parallelismHint</code> is a number that specifies the number of instances of this bolt.</td></tr> |
| <tr><td style="text-align:left"><code>setSpout(String id, IRichSpout spout, Number parallelismHint)</code></td><td style="text-align:left"><code>id</code> is the unique identifier that assigned to a spout, <code>spout</code> is the one previously composed, and <code>parallelismHint</code> is a number that specifying the number of instances of this spout.</td></tr> |
| </tbody> |
| </table> |
| <p>Here's a simple example:</p> |
| <pre><code class="hljs css language-java"> |
| TopologyBuilder builder = <span class="hljs-keyword">new</span> TopologyBuilder(); |
| builder.setSpout(<span class="hljs-string">"word"</span>, <span class="hljs-keyword">new</span> TestWordSpout(), <span class="hljs-number">5</span>); |
| builder.setBolt(<span class="hljs-string">"exclaim"</span>, <span class="hljs-keyword">new</span> ExclamationBolt(), <span class="hljs-number">4</span>); |
| </code></pre> |
| <p>In addition to the component specification, you also need to specify how tuples |
| will be routed between your topology components. There are a few different grouping |
| strategies available:</p> |
| <table> |
| <thead> |
| <tr><th style="text-align:left">Grouping strategy</th><th style="text-align:left">Description</th></tr> |
| </thead> |
| <tbody> |
| <tr><td style="text-align:left">Fields grouping</td><td style="text-align:left">Tuples are transmitted to bolts based on a given field. Tuples with the same field will always go to the same bolt.</td></tr> |
| <tr><td style="text-align:left">Global grouping</td><td style="text-align:left">All tuples are transmitted to a single instance of a bolt with the lowest task id.</td></tr> |
| <tr><td style="text-align:left">Shuffle Grouping</td><td style="text-align:left">Tuples are randomly transmitted to different instances of a bolt.</td></tr> |
| <tr><td style="text-align:left">None grouping</td><td style="text-align:left">Currently, this is the same as shuffle grouping.</td></tr> |
| <tr><td style="text-align:left">All grouping</td><td style="text-align:left">All tuples are transmitted to all instances of a bolt.</td></tr> |
| <tr><td style="text-align:left">Custom grouping</td><td style="text-align:left">User-defined grouping strategy.</td></tr> |
| </tbody> |
| </table> |
| <p>The following snippet is a simple example of specifying shuffle grouping |
| between a <code>word</code> spout and an <code>exclaim</code> bolt.</p> |
| <pre><code class="hljs css language-java"> |
| builder.setBolt(<span class="hljs-string">"exclaim"</span>, <span class="hljs-keyword">new</span> ExclamationBolt(), <span class="hljs-number">4</span>) |
| .shuffleGrouping(<span class="hljs-string">"word"</span>); |
| </code></pre> |
| <p>Once the components and the grouping are specified, the topology can be built.</p> |
| <pre><code class="hljs css language-java">HeronTopology topology = builder.createTopology(); |
| </code></pre> |
| <p>See the <a href="https://github.com/apache/incubator-heron/blob/master/examples/src/java/org/apache/heron/examples/api/ExclamationTopology.java"><code>ExclamationTopology</code></a> for the complete example. More examples can be found in the <a href="https://github.com/apache/incubator-heron/tree/master/examples/src/java/org/apache/heron/examples"><code>examples package</code></a>.</p> |
| <h2><a class="anchor" aria-hidden="true" id="spouts"></a><a href="#spouts" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Spouts</h2> |
| <p>A Heron <strong>spout</strong> is a source of streams, responsible for emitting |
| <a href="../../developers/data-model">tuples</a> into the topology. A spout may, for |
| example, read data from a Kestrel queue or read tweets from the Twitter API and |
| emit tuples to one or more bolts.</p> |
| <p>Information on building spouts can be found in <a href="../../developers/java/spouts">Building |
| Spouts</a>.</p> |
| <h3><a class="anchor" aria-hidden="true" id="implementing-a-spout"></a><a href="#implementing-a-spout" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Implementing a Spout</h3> |
| <p>Spouts must implement the <a href="/api/org/apache/heron/api/spout/ISpout.html"><code>ISpout</code></a> interface.</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">interface</span> <span class="hljs-title">ISpout</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">Serializable</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">open</span><span class="hljs-params">(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector)</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">close</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">activate</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">deactivate</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">nextTuple</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">ack</span><span class="hljs-params">(Object msgId)</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">fail</span><span class="hljs-params">(Object msgId)</span></span>; |
| } |
| </code></pre> |
| <ul> |
| <li><p>The <code>open</code> method is called when the spout is initialized and provides the |
| spout with the executing environment.</p></li> |
| <li><p>The <code>close</code> method is called when the spout is shutdown. There's no guarantee |
| that this method is called due to how the instance is killed.</p></li> |
| <li><p>The <code>activate</code> method is called when the spout is asked to back into active |
| state.</p></li> |
| <li><p>The <code>deactivate</code> method is called when the spout is asked to enter deactive |
| state.</p></li> |
| <li><p>The <code>nextTuple</code> method is used to fetch tuples from input source and emit it |
| to <a href="/api/org/apache/heron/api/bolt/"><code>OutputCollector</code></a>.</p></li> |
| <li><p>The <code>ack</code> method is called when the <code>Tuple</code> with the <code>msgId</code> emitted by this |
| spout is successfully processed.</p></li> |
| <li><p>The <code>fail</code> method is called when the <code>Tuple</code> with the <code>msgId</code> emitted by this |
| spout is not processed successfully.</p></li> |
| </ul> |
| <p>See <a href="https://github.com/apache/incubator-heron/blob/master/examples/src/java/org/apache/heron/examples/api/spout/TestWordSpout.java"><code>TestWordSpout</code></a> for a simple spout example.</p> |
| <p>Instead of implementing the <a href="/api/org/apache/heron/api/spout/ISpout.html"><code>ISpout</code></a> interface directly, you can also implement <a href="/api/org/apache/heron/api/spout/IRichSpout.html"><code>IRichSpout</code></a>.</p> |
| <h2><a class="anchor" aria-hidden="true" id="bolts"></a><a href="#bolts" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Bolts</h2> |
| <p>A Heron <strong>bolt</strong> consumes streams of |
| <a href="guides-data-model">tuples</a> emitted by spouts and performs some |
| set of user-defined processing operations on those tuples, which may include |
| performing complex stream transformations, performing storage operations, |
| aggregating multiple streams into one, emitting tuples to other bolts within the |
| topology, and much more.</p> |
| <h3><a class="anchor" aria-hidden="true" id="implementing-a-bolt"></a><a href="#implementing-a-bolt" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Implementing a Bolt</h3> |
| <p>Spouts must implement the <a href="/api/org/apache/heron/api/spout/ISpout.html"><code>ISpout</code></a> interface.</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">interface</span> <span class="hljs-title">ISpout</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">Serializable</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">open</span><span class="hljs-params">(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector)</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">close</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">activate</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">deactivate</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">nextTuple</span><span class="hljs-params">()</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">ack</span><span class="hljs-params">(Object msgId)</span></span>; |
| <span class="hljs-function"><span class="hljs-keyword">void</span> <span class="hljs-title">fail</span><span class="hljs-params">(Object msgId)</span></span>; |
| } |
| </code></pre> |
| <ul> |
| <li><p>The <code>open</code> method is called when the spout is initialized and provides the |
| spout with the executing environment.</p></li> |
| <li><p>The <code>close</code> method is called when the spout is shutdown. There's no guarantee |
| that this method is called due to how the instance is killed.</p></li> |
| <li><p>The <code>activate</code> method is called when the spout is asked to back into active |
| state.</p></li> |
| <li><p>The <code>deactivate</code> method is called when the spout is asked to enter deactive |
| state.</p></li> |
| <li><p>The <code>nextTuple</code> method is used to fetch tuples from input source and emit it |
| to <a href="/api/org/apache/heron/api/bolt/"><code>OutputCollector</code></a>.</p></li> |
| <li><p>The <code>ack</code> method is called when the <code>Tuple</code> with the <code>msgId</code> emitted by this |
| spout is successfully processed.</p></li> |
| <li><p>The <code>fail</code> method is called when the <code>Tuple</code> with the <code>msgId</code> emitted by this |
| spout is not processed successfully.</p></li> |
| </ul> |
| <p>See <a href="https://github.com/apache/incubator-heron/blob/master/examples/src/java/org/apache/heron/examples/api/spout/TestWordSpout.java"><code>TestWordSpout</code></a> for a simple spout example.</p> |
| <p>Instead of implementing the <a href="/api/org/apache/heron/api/spout/ISpout.html"><code>ISpout</code></a> interface directly, you can also implement <a href="/api/org/apache/heron/api/spout/IRichSpout.html"><code>IRichSpout</code></a>.</p> |
| <h2><a class="anchor" aria-hidden="true" id="applying-delivery-semantics-to-topologies"></a><a href="#applying-delivery-semantics-to-topologies" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Applying delivery semantics to topologies</h2> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">import</span> org.apache.heron.api.Config; |
| |
| Config topologyConfig = <span class="hljs-keyword">new</span> Config(); |
| |
| config.setTopologyReliabilityMode(Config.TopologyReliabilityMode.EFFECTIVELY_ONCE); |
| </code></pre> |
| <p>There are three delivery semantics available corresponding to the three delivery semantics that Heron provides:</p> |
| <ul> |
| <li><code>ATMOST_ONCE</code></li> |
| <li><code>ATLEAST_ONCE</code></li> |
| <li><code>EFFECTIVELY_ONCE</code></li> |
| </ul> |
| <h2><a class="anchor" aria-hidden="true" id="acking"></a><a href="#acking" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Acking</h2> |
| <p>In distributed systems, an <strong>ack</strong> (short for "acknowledgment") is a message that confirms that some action has been taken. In Heron, you can create <a href="#acking-bolts">bolts</a> that emit acks when some desired operation has occurred (for example data has been successfully stored in a database or a message has been successfully produced on a topic in a pub-sub messaging system). Those acks can then be received and acted upon by upstream <a href="#ack-receiving-spouts">spouts</a>.</p> |
| <blockquote> |
| <p>You can see acking at work in a complete Heron topology in <a href="https://github.com/apache/incubator-heron/blob/master/examples/src/java/org/apache/heron/examples/api/AckingTopology.java">this topology</a>.</p> |
| </blockquote> |
| <p>Whereas acking a tuple indicates that some operation has succeeded, the opposite can be indicated when a bolt <a href="#failing">fails</a> a tuple.</p> |
| <h3><a class="anchor" aria-hidden="true" id="acking-bolts"></a><a href="#acking-bolts" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Acking bolts</h3> |
| <p>Each Heron bolt has an <code>OutputCollector</code> that can ack tuples using the <code>ack</code> method. Tuples can be acked inside the <code>execute</code> method that each bolt uses to process incoming tuples. <em>When</em> a bolt acks tuples is up to you. Tuples can be acked immediately upon receipt, after data has been saved to a database, after a message has been successfully published to a pub-sub topic, etc.</p> |
| <p>Here's an example of a bolt that acks tuples when they're successfully processed:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">import</span> org.apache.heron.api.bolt.BaseRichBolt; |
| <span class="hljs-keyword">import</span> org.apache.heron.api.bolt.OutputCollector; |
| <span class="hljs-keyword">import</span> org.apache.heron.api.topology.TopologyContext; |
| |
| <span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">AckingBolt</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">BaseRichBolt</span> </span>{ |
| <span class="hljs-keyword">private</span> OutputCollector outputCollector; |
| |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">prepare</span><span class="hljs-params">(Map conf, TopologyContext context, OutputCollector collector)</span> </span>{ |
| <span class="hljs-keyword">this</span>.outputCollector = collector; |
| } |
| |
| <span class="hljs-function"><span class="hljs-keyword">private</span> <span class="hljs-keyword">void</span> <span class="hljs-title">applyProcessingOperation</span><span class="hljs-params">(Tuple tuple)</span> <span class="hljs-keyword">throws</span> Exception </span>{ |
| <span class="hljs-comment">// Some processing logic for each tuple received by the bolt</span> |
| } |
| |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">execute</span><span class="hljs-params">(Tuple tuple)</span> </span>{ |
| <span class="hljs-keyword">try</span> { |
| applyProcessingOperation(tuple); |
| outputCollector.ack(tuple); |
| } <span class="hljs-keyword">catch</span> (Exception e) { |
| outputCollector.fail(tuple); |
| } |
| } |
| } |
| </code></pre> |
| <p>In this bolt, there's an <code>applyProcessingOperation</code> function that processes each incoming tuple. One of two things can result from this function:</p> |
| <ol> |
| <li>The operation succeeds, in which case the bolt sends an ack. Any upstream spouts---such as a spout like the <code>AckReceivingSpout</code> below---would then receive that ack, along with the message ID that the bolt provides.</li> |
| <li>The operation fails and throws an exception, in which case the tuple is failed rather than acked.</li> |
| </ol> |
| <h3><a class="anchor" aria-hidden="true" id="ack-receiving-spouts"></a><a href="#ack-receiving-spouts" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Ack-receiving spouts</h3> |
| <p>Heron spouts don't emit acks, but they can receive acks when downstream bolts have acked a tuple. In order to receive an ack from downstream bolts, spouts need to do two things:</p> |
| <ol> |
| <li><a href="#specifying-a-message-id">Specify</a> a message ID when they emit tuples using the <code>nextTuple</code> method</li> |
| <li><a href="#specifying-ack-reception-logic">Implement</a> an <code>ack</code> function that specifies what will happen when an ack is received from downstream bolts</li> |
| </ol> |
| <h3><a class="anchor" aria-hidden="true" id="specifying-a-message-id"></a><a href="#specifying-a-message-id" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Specifying a message ID</h3> |
| <p>If you want a spout to receive acks from downstream bolts, the spout needs to specify a message ID every time the spout's <code>SpoutOutputCollector</code> emits a tuple to downstream bolts. Here's an example:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">import</span> org.apache.heron.api.spout.BaseRichSpout; |
| |
| <span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">AckReceivingSpout</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">BaseRichSpout</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">private</span> Object <span class="hljs-title">generateMessageId</span><span class="hljs-params">()</span> </span>{ |
| <span class="hljs-comment">// Some logic to produce a unique ID</span> |
| } |
| |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">nextTuple</span><span class="hljs-params">()</span> </span>{ |
| collector.emit(<span class="hljs-keyword">new</span> Values(someValue), generateMessageId()); |
| } |
| } |
| </code></pre> |
| <p>In this example, each tuple emitted by the spout includes a unique message ID. If no ID is specified, as in the example below, then the spout simply <em>will not receive acks</em>:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">NoAckReceivedSpout</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">BaseRichSpout</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">nextTuple</span><span class="hljs-params">()</span> </span>{ |
| collector.emit(<span class="hljs-keyword">new</span> Values(someValue)); |
| } |
| } |
| </code></pre> |
| <blockquote> |
| <p>When implementing acking logic---as well as <a href="#failing">failing logic</a>---each tuple that is acked/failed <strong>must have a unique ID</strong>. Otherwise, the spout receiving the ack will not be able to identify <em>which</em> tuple has been acked/failed.</p> |
| </blockquote> |
| <p>When specifying an ID for the tuple being emitted, the ID is of type <code>Object</code>, which means that you can serialize to/deserialize from any data type that you'd like. The message ID could thus be a simple <code>String</code> or <code>long</code> or something more complex, like a hash, <code>Map</code>, or POJO.</p> |
| <h3><a class="anchor" aria-hidden="true" id="specifying-ack-reception-logic"></a><a href="#specifying-ack-reception-logic" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Specifying ack reception logic</h3> |
| <p>In order to specify what your spout does when an ack is received, you need to implement an <code>ack</code> function in your spout. That function takes a Java <code>Object</code> containing the tuple's ID, which means that you can potentially serialize the message ID to any type you'd like.</p> |
| <p>In this example, the spout simply logs the message ID:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">AckReceivingSpout</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">BaseRichSpout</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">private</span> Object <span class="hljs-title">generateMessageId</span><span class="hljs-params">()</span> </span>{ |
| <span class="hljs-comment">// Some logic to produce a unique ID</span> |
| } |
| |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">nextTuple</span><span class="hljs-params">()</span> </span>{ |
| collector.emit(<span class="hljs-keyword">new</span> Values(someValue), generateMessageId()); |
| } |
| |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">ack</span><span class="hljs-params">(Object messageId)</span> </span>{ |
| <span class="hljs-comment">// This will simply print the message ID whenever an ack arrives</span> |
| System.out.println((String) messageId); |
| } |
| } |
| </code></pre> |
| <p>In this example, the spout performs a series of actions when receiving the ack:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">AckReceivingSpout</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">BaseRichSpout</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">nextTuple</span><span class="hljs-params">()</span> </span>{ |
| <span class="hljs-keyword">if</span> (someCondition) { |
| String randomHash = <span class="hljs-comment">// Generate a random hash as a message ID</span> |
| collector.emit(<span class="hljs-keyword">new</span> Values(val), randomHash); |
| } |
| } |
| |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">ack</span><span class="hljs-params">(Object messageId)</span> </span>{ |
| saveItemToDatabase(item); |
| publishToPubSubTopic(message); |
| } |
| } |
| </code></pre> |
| <h3><a class="anchor" aria-hidden="true" id="failing"></a><a href="#failing" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Failing</h3> |
| <p><strong>Failing</strong> a tuple is essentially the opposite of acking it, i.e. it indicates that some operation has failed. Bolts can fail tuples by calling the <code>fail</code> method on the <code>OutputCollector</code> rather than <code>ack</code>. Here's an example:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">AckingBolt</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">BaseRichBolt</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">execute</span><span class="hljs-params">(Tuple tuple)</span> </span>{ |
| <span class="hljs-keyword">try</span> { |
| someProcessingOperation(tuple); |
| collector.ack(tuple); |
| } <span class="hljs-keyword">catch</span> (Exception e) { |
| collector.fail(tuple); |
| } |
| } |
| } |
| </code></pre> |
| <p>In this example, an exception-throwing processing operation is attempted. If it succeeds, the tuple is acked; if it fails and an exception is thrown, the tuple is failed.</p> |
| <p>As with acks, spouts can be set up to handle failed tuples by implementing the <code>fail</code> method, which takes the message ID as the argument (just like the <code>ack</code> method). Here's an example:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">public</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">AckReceivingSpout</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">BaseRichSpout</span> </span>{ |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">nextTuple</span><span class="hljs-params">()</span> </span>{ |
| collector.emit(<span class="hljs-keyword">new</span> Values(someValue), someMessageId); |
| } |
| |
| <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">fail</span><span class="hljs-params">(Object messageId)</span> </span>{ |
| <span class="hljs-comment">// Process the messageId</span> |
| } |
| } |
| </code></pre> |
| <p>As with acking, spouts must include a message ID when emitting tuples or else they will not receive fail messages.</p> |
| <h3><a class="anchor" aria-hidden="true" id="acking-failing-and-timeouts"></a><a href="#acking-failing-and-timeouts" aria-hidden="true" class="hash-link"><svg class="hash-link-icon" aria-hidden="true" height="16" version="1.1" viewBox="0 0 16 16" width="16"><path fill-rule="evenodd" d="M4 9h1v1H4c-1.5 0-3-1.69-3-3.5S2.55 3 4 3h4c1.45 0 3 1.69 3 3.5 0 1.41-.91 2.72-2 3.25V8.59c.58-.45 1-1.27 1-2.09C10 5.22 8.98 4 8 4H4c-.98 0-2 1.22-2 2.5S3 9 4 9zm9-3h-1v1h1c1 0 2 1.22 2 2.5S13.98 12 13 12H9c-.98 0-2-1.22-2-2.5 0-.83.42-1.64 1-2.09V6.25c-1.09.53-2 1.84-2 3.25C6 11.31 7.55 13 9 13h4c1.45 0 3-1.69 3-3.5S14.5 6 13 6z"></path></svg></a>Acking, failing, and timeouts</h3> |
| <p>If you're setting up your spouts and bolts to include an ack/fail logic, you can specify that a tuple will automatically be failed if a timeout threshold is reached before the tuple is acked. In this example, all tuples passing through all bolts will be failed if not acked within 10 seconds:</p> |
| <pre><code class="hljs css language-java"><span class="hljs-keyword">import</span> org.apache.heron.api.Config; |
| |
| Config config = <span class="hljs-keyword">new</span> Config(); |
| config.setMessageTimeoutSecs(<span class="hljs-number">10</span>); |
| </code></pre> |
| </span></div></article></div><div class="docs-prevnext"><a class="docs-prev button" href="/docs/next/topology-development-eco-api"><span class="arrow-prev">← </span><span>The ECO API for Java</span></a><a class="docs-next button" href="/docs/next/topology-development-topology-api-python"><span>The Heron Topology API for Python</span><span class="arrow-next"> →</span></a></div></div></div><nav class="onPageNav"><ul class="toc-headings"><li><a href="#getting-started">Getting started</a><ul class="toc-headings"><li><a href="#maven-setup">Maven setup</a></li><li><a href="#writing-your-topology-logic">Writing your topology logic</a></li></ul></li><li><a href="#spouts">Spouts</a><ul class="toc-headings"><li><a href="#implementing-a-spout">Implementing a Spout</a></li></ul></li><li><a href="#bolts">Bolts</a><ul class="toc-headings"><li><a href="#implementing-a-bolt">Implementing a Bolt</a></li></ul></li><li><a href="#applying-delivery-semantics-to-topologies">Applying delivery semantics to topologies</a></li><li><a href="#acking">Acking</a><ul class="toc-headings"><li><a href="#acking-bolts">Acking bolts</a></li><li><a href="#ack-receiving-spouts">Ack-receiving spouts</a></li><li><a href="#specifying-a-message-id">Specifying a message ID</a></li><li><a href="#specifying-ack-reception-logic">Specifying ack reception logic</a></li><li><a href="#failing">Failing</a></li><li><a href="#acking-failing-and-timeouts">Acking, failing, and timeouts</a></li></ul></li></ul></nav></div><footer class="nav-footer" id="footer"><div class="apache-disclaimer">Apache Heron is an effort undergoing incubation at <a target="_blank" href="https://apache.org/">The Apache Software Foundation (ASF)</a> sponsored by the Apache Incubator PMC. Incubation is required of all newly accepted projects until a further review indicates that the infrastructure, communications, and decision making process have stabilized in a manner consistent with other successful ASF projects. While incubation status is not necessarily a reflection of the completeness or stability of the code, it does indicate that the project has yet to be fully endorsed by the ASF.<br/><br/>Apache®, the names of Apache projects, and the feather logo are either <a rel="external" href="https://www.apache.org/foundation/marks/list/">registered trademarks or trademarks</a> of the Apache Software Foundation in the United States and/or other countries.<br/><br/><div class="copyright-box">Copyright © 2023 the Apache Software Foundation, Apache Heron, Heron, |
| Apache, the Apache feather Logo, and the Apache Heron project logo are either registered |
| trademarks or trademarks of the Apache Software Foundation.</div></div><div class="apache-links"><a class="item" rel="external" href="https://incubator.apache.org/">Apache Incubator</a><div><a class="item" rel="external" href="https://www.apache.org/">About the ASF</a></div><div><a class="item" rel="external" href="https://www.apache.org/events/current-event">Events</a></div><div><a class="item" rel="external" href="https://www.apache.org/foundation/thanks.html">Thanks</a></div><div><a class="item" rel="external" href="https://www.apache.org/foundation/sponsorship.html">Become a Sponsor</a></div><div><a class="item" rel="external" href="https://www.apache.org/security/">Security</a></div><div><a class="item" rel="external" href="https://www.apache.org/licenses/">License</a></div></div></footer></div><script>window.twttr=(function(d,s, id){var js,fjs=d.getElementsByTagName(s)[0],t=window.twttr||{};if(d.getElementById(id))return t;js=d.createElement(s);js.id=id;js.src='https://platform.twitter.com/widgets.js';fjs.parentNode.insertBefore(js, fjs);t._e = [];t.ready = function(f) {t._e.push(f);};return t;}(document, 'script', 'twitter-wjs'));</script></body></html> |
