You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by th...@apache.org on 2017/03/27 15:19:10 UTC
[09/19] apex-malhar git commit: An application that demonstrates
throttling of the input operator output rate when the downsteam operators are
slower Fixed throttler to send requests once per state change instead of
repeatedly
An application that demonstrates throttling of the input operator output rate when the downsteam operators are slower
Fixed throttler to send requests once per state change instead of repeatedly
Project: http://git-wip-us.apache.org/repos/asf/apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/apex-malhar/commit/172b20a2
Tree: http://git-wip-us.apache.org/repos/asf/apex-malhar/tree/172b20a2
Diff: http://git-wip-us.apache.org/repos/asf/apex-malhar/diff/172b20a2
Branch: refs/heads/master
Commit: 172b20a2c0bb6ccbd584b34aff41779da9d7acfb
Parents: dd509f2
Author: Pramod Immaneni <pr...@datatorrent.com>
Authored: Sun Nov 6 01:33:05 2016 -0700
Committer: Lakshmi Prasanna Velineni <la...@datatorrent.com>
Committed: Sun Mar 26 11:43:48 2017 -0700
----------------------------------------------------------------------
examples/throttle/README.md | 8 +
.../throttle/XmlJavadocCommentsExtractor.xsl | 44 +++
examples/throttle/pom.xml | 274 +++++++++++++++++++
examples/throttle/src/assemble/appPackage.xml | 43 +++
.../examples/throttle/Application.java | 51 ++++
.../examples/throttle/PassThroughOperator.java | 20 ++
.../throttle/RandomNumberGenerator.java | 64 +++++
.../examples/throttle/SlowDevNullOperator.java | 35 +++
.../throttle/ThrottlingStatsListener.java | 150 ++++++++++
.../src/main/resources/META-INF/properties.xml | 20 ++
.../examples/throttle/ApplicationTest.java | 37 +++
.../src/test/resources/log4j.properties | 22 ++
12 files changed, 768 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/README.md
----------------------------------------------------------------------
diff --git a/examples/throttle/README.md b/examples/throttle/README.md
new file mode 100644
index 0000000..7ed4670
--- /dev/null
+++ b/examples/throttle/README.md
@@ -0,0 +1,8 @@
+Application shows you how to throttle input operators in the application when the downstream
+operators are slower.
+
+It uses a combination of stats listener and operator request to achieve this. The throttler is
+a stats listener that is registered with the operators and when it notices that the window gap
+between operators is widening and crosses a configured threshold, it sends a request to the
+input operator to slow down and coversely when the gap falls below the threshold it requests
+the input operator to go back to its normal speed.
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/XmlJavadocCommentsExtractor.xsl
----------------------------------------------------------------------
diff --git a/examples/throttle/XmlJavadocCommentsExtractor.xsl b/examples/throttle/XmlJavadocCommentsExtractor.xsl
new file mode 100644
index 0000000..08075a9
--- /dev/null
+++ b/examples/throttle/XmlJavadocCommentsExtractor.xsl
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed 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.
+
+-->
+
+<!--
+ Document : XmlJavadocCommentsExtractor.xsl
+ Created on : September 16, 2014, 11:30 AM
+ Description:
+ The transformation strips off all information except for comments and tags from xml javadoc generated by xml-doclet.
+-->
+
+<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform" version="1.0">
+ <xsl:output method="xml" standalone="yes"/>
+
+ <!-- copy xml by selecting only the following nodes, attributes and text -->
+ <xsl:template match="node()|text()|@*">
+ <xsl:copy>
+ <xsl:apply-templates select="root|package|class|interface|method|field|type|comment|tag|text()|@name|@qualified|@text"/>
+ </xsl:copy>
+ </xsl:template>
+
+ <!-- Strip off the following paths from the selected xml -->
+ <xsl:template match="//root/package/interface/interface
+ |//root/package/interface/method/@qualified
+ |//root/package/class/interface
+ |//root/package/class/class
+ |//root/package/class/method/@qualified
+ |//root/package/class/field/@qualified" />
+
+ <xsl:strip-space elements="*"/>
+</xsl:stylesheet>
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/pom.xml
----------------------------------------------------------------------
diff --git a/examples/throttle/pom.xml b/examples/throttle/pom.xml
new file mode 100644
index 0000000..02fe22e
--- /dev/null
+++ b/examples/throttle/pom.xml
@@ -0,0 +1,274 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>com.datatorrent.examples</groupId>
+ <version>1.0-SNAPSHOT</version>
+ <artifactId>throttle</artifactId>
+ <packaging>jar</packaging>
+
+ <!-- change these to the appropriate values -->
+ <name>Throttle Application</name>
+ <description>Application demonstrating throttling input when downstream is slower</description>
+
+ <properties>
+ <!-- change this if you desire to use a different version of Apex Core -->
+ <apex.version>3.5.0</apex.version>
+ <apex.apppackage.classpath>lib/*.jar</apex.apppackage.classpath>
+ </properties>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-eclipse-plugin</artifactId>
+ <version>2.9</version>
+ <configuration>
+ <downloadSources>true</downloadSources>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.3</version>
+ <configuration>
+ <encoding>UTF-8</encoding>
+ <source>1.7</source>
+ <target>1.7</target>
+ <debug>true</debug>
+ <optimize>false</optimize>
+ <showDeprecation>true</showDeprecation>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <version>2.8</version>
+ <executions>
+ <execution>
+ <id>copy-dependencies</id>
+ <phase>prepare-package</phase>
+ <goals>
+ <goal>copy-dependencies</goal>
+ </goals>
+ <configuration>
+ <outputDirectory>target/deps</outputDirectory>
+ <includeScope>runtime</includeScope>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>app-package-assembly</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <finalName>${project.artifactId}-${project.version}-apexapp</finalName>
+ <appendAssemblyId>false</appendAssemblyId>
+ <descriptors>
+ <descriptor>src/assemble/appPackage.xml</descriptor>
+ </descriptors>
+ <archiverConfig>
+ <defaultDirectoryMode>0755</defaultDirectoryMode>
+ </archiverConfig>
+ <archive>
+ <manifestEntries>
+ <Class-Path>${apex.apppackage.classpath}</Class-Path>
+ <DT-Engine-Version>${apex.version}</DT-Engine-Version>
+ <DT-App-Package-Group-Id>${project.groupId}</DT-App-Package-Group-Id>
+ <DT-App-Package-Name>${project.artifactId}</DT-App-Package-Name>
+ <DT-App-Package-Version>${project.version}</DT-App-Package-Version>
+ <DT-App-Package-Display-Name>${project.name}</DT-App-Package-Display-Name>
+ <DT-App-Package-Description>${project.description}</DT-App-Package-Description>
+ </manifestEntries>
+ </archive>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <version>1.7</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <configuration>
+ <target>
+ <move file="${project.build.directory}/${project.artifactId}-${project.version}-apexapp.jar"
+ tofile="${project.build.directory}/${project.artifactId}-${project.version}.apa" />
+ </target>
+ </configuration>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ </execution>
+ <execution>
+ <!-- create resource directory for xml javadoc-->
+ <id>createJavadocDirectory</id>
+ <phase>generate-resources</phase>
+ <configuration>
+ <tasks>
+ <delete dir="${project.build.directory}/generated-resources/xml-javadoc"/>
+ <mkdir dir="${project.build.directory}/generated-resources/xml-javadoc"/>
+ </tasks>
+ </configuration>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>1.9.1</version>
+ <executions>
+ <execution>
+ <id>attach-artifacts</id>
+ <phase>package</phase>
+ <goals>
+ <goal>attach-artifact</goal>
+ </goals>
+ <configuration>
+ <artifacts>
+ <artifact>
+ <file>target/${project.artifactId}-${project.version}.apa</file>
+ <type>apa</type>
+ </artifact>
+ </artifacts>
+ <skipAttach>false</skipAttach>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- generate javdoc -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <executions>
+ <!-- generate xml javadoc -->
+ <execution>
+ <id>xml-doclet</id>
+ <phase>generate-resources</phase>
+ <goals>
+ <goal>javadoc</goal>
+ </goals>
+ <configuration>
+ <doclet>com.github.markusbernhardt.xmldoclet.XmlDoclet</doclet>
+ <additionalparam>-d ${project.build.directory}/generated-resources/xml-javadoc -filename ${project.artifactId}-${project.version}-javadoc.xml</additionalparam>
+ <useStandardDocletOptions>false</useStandardDocletOptions>
+ <docletArtifact>
+ <groupId>com.github.markusbernhardt</groupId>
+ <artifactId>xml-doclet</artifactId>
+ <version>1.0.4</version>
+ </docletArtifact>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!-- Transform xml javadoc to stripped down version containing only class/interface comments and tags-->
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>xml-maven-plugin</artifactId>
+ <version>1.0</version>
+ <executions>
+ <execution>
+ <id>transform-xmljavadoc</id>
+ <phase>generate-resources</phase>
+ <goals>
+ <goal>transform</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <transformationSets>
+ <transformationSet>
+ <dir>${project.build.directory}/generated-resources/xml-javadoc</dir>
+ <includes>
+ <include>${project.artifactId}-${project.version}-javadoc.xml</include>
+ </includes>
+ <stylesheet>XmlJavadocCommentsExtractor.xsl</stylesheet>
+ <outputDir>${project.build.directory}/generated-resources/xml-javadoc</outputDir>
+ </transformationSet>
+ </transformationSets>
+ </configuration>
+ </plugin>
+ <!-- copy xml javadoc to class jar -->
+ <plugin>
+ <artifactId>maven-resources-plugin</artifactId>
+ <version>2.6</version>
+ <executions>
+ <execution>
+ <id>copy-resources</id>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>copy-resources</goal>
+ </goals>
+ <configuration>
+ <outputDirectory>${basedir}/target/classes</outputDirectory>
+ <resources>
+ <resource>
+ <directory>${project.build.directory}/generated-resources/xml-javadoc</directory>
+ <includes>
+ <include>${project.artifactId}-${project.version}-javadoc.xml</include>
+ </includes>
+ <filtering>true</filtering>
+ </resource>
+ </resources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ </plugins>
+
+ </build>
+
+ <dependencies>
+ <!-- add your dependencies here -->
+ <dependency>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>malhar-library</artifactId>
+ <version>3.6.0</version>
+ <!--
+ If you know that your application does not need transitive dependencies pulled in by malhar-library,
+ uncomment the following to reduce the size of your app package.
+ -->
+ <!--
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ -->
+ </dependency>
+ <dependency>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex-common</artifactId>
+ <version>${apex.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.10</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.apex</groupId>
+ <artifactId>apex-engine</artifactId>
+ <version>${apex.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/assemble/appPackage.xml
----------------------------------------------------------------------
diff --git a/examples/throttle/src/assemble/appPackage.xml b/examples/throttle/src/assemble/appPackage.xml
new file mode 100644
index 0000000..7ad071c
--- /dev/null
+++ b/examples/throttle/src/assemble/appPackage.xml
@@ -0,0 +1,43 @@
+<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd">
+ <id>appPackage</id>
+ <formats>
+ <format>jar</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${basedir}/target/</directory>
+ <outputDirectory>/app</outputDirectory>
+ <includes>
+ <include>${project.artifactId}-${project.version}.jar</include>
+ </includes>
+ </fileSet>
+ <fileSet>
+ <directory>${basedir}/target/deps</directory>
+ <outputDirectory>/lib</outputDirectory>
+ </fileSet>
+ <fileSet>
+ <directory>${basedir}/src/site/conf</directory>
+ <outputDirectory>/conf</outputDirectory>
+ <includes>
+ <include>*.xml</include>
+ </includes>
+ </fileSet>
+ <fileSet>
+ <directory>${basedir}/src/main/resources/META-INF</directory>
+ <outputDirectory>/META-INF</outputDirectory>
+ </fileSet>
+ <fileSet>
+ <directory>${basedir}/src/main/resources/app</directory>
+ <outputDirectory>/app</outputDirectory>
+ </fileSet>
+ <fileSet>
+ <directory>${basedir}/src/main/resources/resources</directory>
+ <outputDirectory>/resources</outputDirectory>
+ </fileSet>
+ </fileSets>
+
+</assembly>
+
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/main/java/com/datatorrent/examples/throttle/Application.java
----------------------------------------------------------------------
diff --git a/examples/throttle/src/main/java/com/datatorrent/examples/throttle/Application.java b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/Application.java
new file mode 100644
index 0000000..d789dbe
--- /dev/null
+++ b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/Application.java
@@ -0,0 +1,51 @@
+/**
+ * Put your copyright and license info here.
+ */
+package com.datatorrent.examples.throttle;
+
+import java.util.Collection;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.StatsListener;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+
+@ApplicationAnnotation(name="ThrottleApplication")
+public class Application implements StreamingApplication
+{
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ // Creating an example application with three operators
+ // The last operator is slowing down the DAG
+ // With the use of the stats listener the input operator is slowed when the window difference crosses a threshold
+
+ RandomNumberGenerator randomGenerator = dag.addOperator("RandomGenerator", RandomNumberGenerator.class);
+ PassThroughOperator<Double> passThrough = dag.addOperator("PassThrough", PassThroughOperator.class);
+ SlowDevNullOperator<Double> devNull = dag.addOperator("SlowNull", SlowDevNullOperator.class);
+
+ // Important to use the same stats listener object for all operators so that we can centrally collect stats and make
+ // the decision
+ StatsListener statsListener = new ThrottlingStatsListener();
+ Collection<StatsListener> statsListeners = Lists.newArrayList(statsListener);
+ dag.setAttribute(randomGenerator, Context.OperatorContext.STATS_LISTENERS, statsListeners);
+ dag.setAttribute(passThrough, Context.OperatorContext.STATS_LISTENERS, statsListeners);
+ dag.setAttribute(devNull, Context.OperatorContext.STATS_LISTENERS, statsListeners);
+
+ // Increase timeout for the slow operator, this specifies the maximum timeout for an operator to process a window
+ // It is specified in number of windows, since 1 window is 500ms, 30 minutes is 30 * 60 * 2 = 3600 windows
+ dag.setAttribute(devNull, Context.OperatorContext.TIMEOUT_WINDOW_COUNT, 3600);
+
+ // If there are unifiers that are slow then set timeout for them
+ // dag.setUnifierAttribute(passThrough.output, Context.OperatorContext.TIMEOUT_WINDOW_COUNT, 3600);
+
+ dag.addStream("randomData", randomGenerator.out, passThrough.input);
+ dag.addStream("passData", passThrough.output, devNull.input);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/main/java/com/datatorrent/examples/throttle/PassThroughOperator.java
----------------------------------------------------------------------
diff --git a/examples/throttle/src/main/java/com/datatorrent/examples/throttle/PassThroughOperator.java b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/PassThroughOperator.java
new file mode 100644
index 0000000..b4630df
--- /dev/null
+++ b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/PassThroughOperator.java
@@ -0,0 +1,20 @@
+package com.datatorrent.examples.throttle;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * Created by pramod on 9/27/16.
+ */
+public class PassThroughOperator<T> extends BaseOperator {
+
+ public transient final DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t) {
+ output.emit(t);
+ }
+ };
+
+ public transient final DefaultOutputPort<T> output = new DefaultOutputPort<>();
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/main/java/com/datatorrent/examples/throttle/RandomNumberGenerator.java
----------------------------------------------------------------------
diff --git a/examples/throttle/src/main/java/com/datatorrent/examples/throttle/RandomNumberGenerator.java b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/RandomNumberGenerator.java
new file mode 100644
index 0000000..ea57b6d
--- /dev/null
+++ b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/RandomNumberGenerator.java
@@ -0,0 +1,64 @@
+/**
+ * Put your copyright and license info here.
+ */
+package com.datatorrent.examples.throttle;
+
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.common.util.BaseOperator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is a simple operator that emits random number.
+ */
+public class RandomNumberGenerator extends BaseOperator implements InputOperator
+{
+ private int numTuples = 1000;
+ private int origNumTuples = numTuples;
+ private transient int count = 0;
+
+ private static final Logger logger = LoggerFactory.getLogger(RandomNumberGenerator.class);
+
+ public final transient DefaultOutputPort<Double> out = new DefaultOutputPort<Double>();
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ count = 0;
+ }
+
+ @Override
+ public void emitTuples()
+ {
+ if (count++ < numTuples) {
+ out.emit(Math.random());
+ }
+ }
+
+ // Simple suspend and
+ public void suspend() {
+ logger.debug("Slowing down");
+ numTuples = 0;
+ }
+
+ public void normal() {
+ logger.debug("Normal");
+ numTuples = origNumTuples;
+ }
+
+ public int getNumTuples()
+ {
+ return numTuples;
+ }
+
+ /**
+ * Sets the number of tuples to be emitted every window.
+ * @param numTuples number of tuples
+ */
+ public void setNumTuples(int numTuples)
+ {
+ this.numTuples = numTuples;
+ this.origNumTuples = numTuples;
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/main/java/com/datatorrent/examples/throttle/SlowDevNullOperator.java
----------------------------------------------------------------------
diff --git a/examples/throttle/src/main/java/com/datatorrent/examples/throttle/SlowDevNullOperator.java b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/SlowDevNullOperator.java
new file mode 100644
index 0000000..7d1451d
--- /dev/null
+++ b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/SlowDevNullOperator.java
@@ -0,0 +1,35 @@
+package com.datatorrent.examples.throttle;
+
+import com.google.common.base.Throwables;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * Created by pramod on 9/27/16.
+ */
+public class SlowDevNullOperator<T> extends BaseOperator {
+
+ // Modify sleep time dynamically while app is running to increase and decrease sleep time
+ long sleepTime = 1;
+
+ public transient final DefaultInputPort<T> input = new DefaultInputPort<T>() {
+ @Override
+ public void process(T t) {
+ // Introduce an artificial delay for every tuple
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {
+ throw Throwables.propagate(e);
+ }
+ }
+ };
+
+ public long getSleepTime() {
+ return sleepTime;
+ }
+
+ public void setSleepTime(long sleepTime) {
+ this.sleepTime = sleepTime;
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/main/java/com/datatorrent/examples/throttle/ThrottlingStatsListener.java
----------------------------------------------------------------------
diff --git a/examples/throttle/src/main/java/com/datatorrent/examples/throttle/ThrottlingStatsListener.java b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/ThrottlingStatsListener.java
new file mode 100644
index 0000000..46e2e0e
--- /dev/null
+++ b/examples/throttle/src/main/java/com/datatorrent/examples/throttle/ThrottlingStatsListener.java
@@ -0,0 +1,150 @@
+package com.datatorrent.examples.throttle;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.StatsListener;
+
+/**
+ * Created by pramod on 9/27/16.
+ */
+public class ThrottlingStatsListener implements StatsListener, Serializable {
+
+ private static final Logger logger = LoggerFactory.getLogger(ThrottlingStatsListener.class);
+
+ // Slowdown input if the window difference between operators increases beyond this value
+ long maxThreshold = 100;
+ // restore input operator to normal speed if the window difference falls below this threshold
+ long minThreshold = 100;
+
+ Map<Integer, ThrottleState> throttleStates = Maps.newHashMap();
+
+ static class ThrottleState {
+ // The current state of the operator, normal or throttled
+ boolean normal = true;
+ //The latest window id for which stats were received for the operator
+ long currentWindowId;
+ }
+
+ // This method runs on the app master side and is called whenever new stats are received from the operators
+ @Override
+ public Response processStats(BatchedOperatorStats batchedOperatorStats)
+ {
+ Response response = new Response();
+ int operatorId = batchedOperatorStats.getOperatorId();
+
+ ThrottleState throttleState = throttleStates.get(operatorId);
+ if (throttleState == null) {
+ throttleState = new ThrottleState();
+ throttleStates.put(operatorId, throttleState);
+ }
+
+ long windowId = batchedOperatorStats.getCurrentWindowId();
+ throttleState.currentWindowId = windowId;
+
+ // Find min and max window to compute difference
+ long minWindow = Long.MAX_VALUE;
+ long maxWindow = Long.MIN_VALUE;
+ for (ThrottleState state : throttleStates.values()) {
+ if (state.currentWindowId < minWindow) minWindow = state.currentWindowId;
+ if (state.currentWindowId > maxWindow) maxWindow = state.currentWindowId;
+ }
+ logger.debug("Operator {} min window {} max window {}", operatorId, minWindow, maxWindow);
+
+ if (throttleState.normal && ((maxWindow - minWindow) > maxThreshold)) {
+ // Send request to operator to slow down
+ logger.info("Sending suspend request");
+ List<OperatorRequest> operatorRequests = new ArrayList<OperatorRequest>();
+ operatorRequests.add(new InputSlowdownRequest());
+ response.operatorRequests = operatorRequests;
+ //logger.info("Setting suspend");
+ throttleState.normal = false;
+ } else if (!throttleState.normal && ((maxWindow - minWindow) <= minThreshold)) {
+ // Send request to operator to get back to normal
+ logger.info("Sending normal request");
+ List<OperatorRequest> operatorRequests = new ArrayList<OperatorRequest>();
+ operatorRequests.add(new InputNormalRequest());
+ response.operatorRequests = operatorRequests;
+ //logger.info("Setting normal");
+ throttleState.normal = true;
+ }
+
+ return response;
+ }
+
+ // This runs on the operator side
+ public static class InputSlowdownRequest implements OperatorRequest, Serializable
+ {
+ private static final Logger logger = LoggerFactory.getLogger(InputSlowdownRequest.class);
+
+ @Override
+ public OperatorResponse execute(Operator operator, int operatorId, long windowId) throws IOException
+ {
+ logger.debug("Received slowdown operator {} operatorId {} windowId {}", operator, operatorId, windowId);
+ if (operator instanceof RandomNumberGenerator) {
+ RandomNumberGenerator generator = (RandomNumberGenerator)operator;
+ generator.suspend();
+ }
+ return new InputOperatorResponse();
+ }
+ }
+
+ public static class InputNormalRequest implements OperatorRequest, Serializable
+ {
+ private static final Logger logger = LoggerFactory.getLogger(InputNormalRequest.class);
+
+ @Override
+ public OperatorResponse execute(Operator operator, int operatorId, long windowId) throws IOException
+ {
+ logger.debug("Received normal operator {} operatorId {} windowId {}", operator, operatorId, windowId);
+ if (operator instanceof RandomNumberGenerator) {
+ RandomNumberGenerator generator = (RandomNumberGenerator)operator;
+ generator.normal();
+ }
+ return new InputOperatorResponse();
+ }
+ }
+
+ public static class InputOperatorResponse implements OperatorResponse, Serializable
+ {
+
+ @Override
+ public Object getResponseId() {
+ return 1;
+ }
+
+ @Override
+ public Object getResponse() {
+ return "";
+ }
+ }
+
+ public long getMaxThreshold()
+ {
+ return maxThreshold;
+ }
+
+ public void setMaxThreshold(long maxThreshold)
+ {
+ this.maxThreshold = maxThreshold;
+ }
+
+ public long getMinThreshold()
+ {
+ return minThreshold;
+ }
+
+ public void setMinThreshold(long minThreshold)
+ {
+ this.minThreshold = minThreshold;
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/main/resources/META-INF/properties.xml
----------------------------------------------------------------------
diff --git a/examples/throttle/src/main/resources/META-INF/properties.xml b/examples/throttle/src/main/resources/META-INF/properties.xml
new file mode 100644
index 0000000..a6ddc4c
--- /dev/null
+++ b/examples/throttle/src/main/resources/META-INF/properties.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0"?>
+<configuration>
+ <!--
+ <property>
+ <name>dt.application.{appName}.operator.{opName}.prop.{propName}</name>
+ <value>some-default-value (if value is not specified, it is required from the user or custom config when launching)</value>
+ </property>
+ -->
+ <!-- memory assigned to app master
+ <property>
+ <name>dt.attr.MASTER_MEMORY_MB</name>
+ <value>1024</value>
+ </property>
+ -->
+ <property>
+ <name>dt.application.MyFirstApplication.operator.randomGenerator.prop.numTuples</name>
+ <value>1000</value>
+ </property>
+</configuration>
+
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/test/java/com/datatorrent/examples/throttle/ApplicationTest.java
----------------------------------------------------------------------
diff --git a/examples/throttle/src/test/java/com/datatorrent/examples/throttle/ApplicationTest.java b/examples/throttle/src/test/java/com/datatorrent/examples/throttle/ApplicationTest.java
new file mode 100644
index 0000000..02a96ba
--- /dev/null
+++ b/examples/throttle/src/test/java/com/datatorrent/examples/throttle/ApplicationTest.java
@@ -0,0 +1,37 @@
+/**
+ * Put your copyright and license info here.
+ */
+package com.datatorrent.examples.throttle;
+
+import java.io.IOException;
+
+import javax.validation.ConstraintViolationException;
+
+import org.junit.Assert;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+import com.datatorrent.api.LocalMode;
+import com.datatorrent.examples.throttle.Application;
+
+/**
+ * Test the DAG declaration in local mode.
+ */
+public class ApplicationTest {
+
+ @Test
+ public void testApplication() throws IOException, Exception {
+ try {
+ LocalMode lma = LocalMode.newInstance();
+ Configuration conf = new Configuration(false);
+ conf.addResource(this.getClass().getResourceAsStream("/META-INF/properties.xml"));
+ lma.prepareDAG(new Application(), conf);
+ LocalMode.Controller lc = lma.getController();
+ lc.run(10000); // runs for 10 seconds and quits
+ } catch (ConstraintViolationException e) {
+ Assert.fail("constraint violations: " + e.getConstraintViolations());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/172b20a2/examples/throttle/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/examples/throttle/src/test/resources/log4j.properties b/examples/throttle/src/test/resources/log4j.properties
new file mode 100644
index 0000000..98544e8
--- /dev/null
+++ b/examples/throttle/src/test/resources/log4j.properties
@@ -0,0 +1,22 @@
+log4j.rootLogger=DEBUG,CONSOLE
+
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [%t] %-5p %c{2} %M - %m%n
+
+log4j.appender.RFA=org.apache.log4j.RollingFileAppender
+log4j.appender.RFA.layout=org.apache.log4j.PatternLayout
+log4j.appender.RFA.layout.ConversionPattern=%d{ISO8601} [%t] %-5p %c{2} %M - %m%n
+log4j.appender.RFA.File=/tmp/app.log
+
+# to enable, add SYSLOG to rootLogger
+log4j.appender.SYSLOG=org.apache.log4j.net.SyslogAppender
+log4j.appender.SYSLOG.syslogHost=127.0.0.1
+log4j.appender.SYSLOG.layout=org.apache.log4j.PatternLayout
+log4j.appender.SYSLOG.layout.conversionPattern=${dt.cid} %-5p [%t] %c{2} %x - %m%n
+log4j.appender.SYSLOG.Facility=LOCAL1
+
+#log4j.logger.org.apache.commons.beanutils=warn
+log4j.logger.com.datatorrent=debug
+log4j.logger.org.apache.apex=debug
+log4j.logger.org=info