Buffer limit (#25)

* Create pulsar-functions module (#1)

* Create pulsar-functions module

* rename `sdk` package to `api`

* Added the first cut of the Java interface for Pulsar functions (#2)

* Added a way to limit the number of outstanding tuples read but still not executed

pulsar-functions/runtime/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java

@@ -98,7 +98,8 @@ void run() throws Exception {
 
             LimitsConfig limitsConfig = new LimitsConfig(
                 60000,   // 60 seconds
-                1024        // 1GB
+                1024,       // 1GB
+                1024   // 1024 outstanding tuples
             );
 
             Spawner spawner = Spawner.createSpawner(

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ThreadFunctionContainer.java

@@ -63,11 +63,11 @@ class Payload {
     private LinkedBlockingQueue<Payload> queue;
     private String id;
 
-    ThreadFunctionContainer(JavaInstanceConfig instanceConfig,
+    ThreadFunctionContainer(JavaInstanceConfig instanceConfig, int maxBufferedTuples,
                             FunctionCacheManager fnCache, ThreadGroup threadGroup) {
         this.javaInstanceConfig = instanceConfig;
         this.fnCache = fnCache;
-        this.queue = new LinkedBlockingQueue<>();
+        this.queue = new LinkedBlockingQueue<>(maxBufferedTuples);
         this.id = "fn-" + instanceConfig.getFunctionConfig().getName() + "-instance-" + instanceConfig.getInstanceId();
         this.fnThread = new Thread(threadGroup,
                 new Runnable() {

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ThreadFunctionContainerFactory.java

@@ -30,17 +30,20 @@ public class ThreadFunctionContainerFactory implements FunctionContainerFactory
 
     private final ThreadGroup threadGroup;
     protected final FunctionCacheManager fnCache;
+    private int maxBufferedTuples;
 
-    public ThreadFunctionContainerFactory() {
+    public ThreadFunctionContainerFactory(int maxBufferedTuples) {
         this.fnCache = new FunctionCacheManagerImpl();
         this.threadGroup = new ThreadGroup(
             "Pulsar Function Container Threads");
+        this.maxBufferedTuples = maxBufferedTuples;
     }
 
     @Override
     public ThreadFunctionContainer createContainer(JavaInstanceConfig instanceConfig) {
         return new ThreadFunctionContainer(
             instanceConfig,
+            maxBufferedTuples,
             fnCache,
             threadGroup);
     }

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/spawner/LimitsConfig.java

@@ -32,4 +32,5 @@
 public class LimitsConfig {
     private int timeBudgetInMs;
     private int maxMemory;
+    private int maxBufferedTuples;
 }

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/spawner/Spawner.java

@@ -61,7 +61,7 @@ public Spawner(LimitsConfig limitsConfig, AssignmentInfo assignmentInfo, String
         this.limitsConfig = limitsConfig;
         this.assignmentInfo = assignmentInfo;
         this.pulsarBrokerRootUrl = pulsarBrokerRootUrl;
-        this.threadFunctionContainerFactory = new ThreadFunctionContainerFactory();
+        this.threadFunctionContainerFactory = new ThreadFunctionContainerFactory(limitsConfig.getMaxBufferedTuples());
     }
 
     public void start() throws Exception {

pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/container/ThreadFunctionContainerTest.java

@@ -55,7 +55,7 @@ public ThreadFunctionContainerTest() {
         URL jarUrl = getClass().getClassLoader().getResource("multifunction.jar");
         this.jarFiles = Lists.newArrayList(jarUrl.getPath());
         this.classpaths = Collections.emptyList();
-        this.factory = new ThreadFunctionContainerFactory();
+        this.factory = new ThreadFunctionContainerFactory(1024);
     }
 
     @After