/**
 * 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.
 */

package org.apache.pulsar.functions.runtime;

import java.util.concurrent.CompletableFuture;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManager;
import org.apache.pulsar.functions.instance.JavaInstanceRunnable;
import org.apache.pulsar.functions.utils.FunctionConfigUtils;

/**
 * A function container implemented using java thread.
 */
@Slf4j
class ThreadRuntime implements Runtime {

    // The thread that invokes the function
    @Getter
    private final Thread fnThread;

    @Getter
    private InstanceConfig instanceConfig;
    private JavaInstanceRunnable javaInstanceRunnable;
    private Exception startupException;

    ThreadRuntime(InstanceConfig instanceConfig,
                  FunctionCacheManager fnCache,
                  ThreadGroup threadGroup,
                  String jarFile,
                  PulsarClient pulsarClient,
                  String stateStorageServiceUrl) {
        this.instanceConfig = instanceConfig;
        if (instanceConfig.getFunctionConfig().getRuntime() != Function.FunctionConfig.Runtime.JAVA) {
            throw new RuntimeException("Thread Container only supports Java Runtime");
        }
        this.javaInstanceRunnable = new JavaInstanceRunnable(
            instanceConfig,
            fnCache,
            jarFile,
            pulsarClient,
            stateStorageServiceUrl);
        this.fnThread = new Thread(threadGroup, javaInstanceRunnable,
                FunctionConfigUtils.getFullyQualifiedName(instanceConfig.getFunctionConfig()));
    }

    /**
     * The core logic that initialize the thread container and executes the function
     */
    @Override
    public void start() {
        log.info("ThreadContainer starting function with instance config {}", instanceConfig);
        startupException = null;
        this.fnThread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            @Override
            public void uncaughtException(Thread t, Throwable e) {
                startupException = new Exception(e);
            }
        });
        this.fnThread.start();
    }

    @Override
    public void join() throws Exception {
        this.fnThread.join();
    }

    @Override
    public void stop() {
        // interrupt the instance thread
        fnThread.interrupt();
        javaInstanceRunnable.close();
        try {
            fnThread.join();
        } catch (InterruptedException e) {
            // ignore this
        }
    }

    @Override
    public CompletableFuture<FunctionStatus> getFunctionStatus() {
        FunctionStatus.Builder functionStatusBuilder = javaInstanceRunnable.getFunctionStatus();
        if (javaInstanceRunnable.getFailureException() != null) {
            functionStatusBuilder.setRunning(false);
            functionStatusBuilder.setFailureException(javaInstanceRunnable.getFailureException().getMessage());
        } else {
            functionStatusBuilder.setRunning(true);
        }
        return CompletableFuture.completedFuture(functionStatusBuilder.build());
    }

    @Override
    public CompletableFuture<InstanceCommunication.MetricsData> getAndResetMetrics() {
        return CompletableFuture.completedFuture(javaInstanceRunnable.getAndResetMetrics());
    }

    @Override
    public boolean isAlive() {
        return this.fnThread.isAlive();
    }

    @Override
    public Exception getDeathException() {
        if (isAlive()) return null;
        else return startupException;
    }
}

@@ -1619,28 +1619,16 @@
 unction%0A
-    @Getter%0A
     priv
@@ -1635,14 +1635,8 @@
 ate 
-final 
 Thre
@@ -1798,16 +1798,53 @@
 ception;
+%0A    private ThreadGroup threadGroup;
 %0A%0A    Th
@@ -2565,159 +2565,33 @@
 his.
-fnThread = new Thread(threadGroup, javaInstanceRunnable,%0A                FunctionConfigUtils.getFullyQualifiedName(instanceConfig.getFunctionConfig()))
+threadGroup = threadGroup
 ;%0A  
@@ -2860,24 +2860,190 @@
 ion = null;%0A
+        this.fnThread = new Thread(threadGroup, javaInstanceRunnable,%0A                FunctionConfigUtils.getFullyQualifiedName(instanceConfig.getFunctionConfig()));%0A
         this
@@ -3382,16 +3382,57 @@
 ption %7B%0A
+        if (this.fnThread != null) %7B%0A    
         
@@ -3449,24 +3449,34 @@
 ead.join();%0A
+        %7D%0A
     %7D%0A%0A    @
@@ -3505,24 +3505,60 @@
 id stop() %7B%0A
+        if (fnThread != null) %7B%0A    
         // i
@@ -3586,32 +3586,36 @@
  thread%0A        
+    
 fnThread.interru
@@ -3628,16 +3628,20 @@
         
+   
+ 
 javaInst
@@ -3670,16 +3670,20 @@
         
+    
 try %7B%0A  
@@ -3684,32 +3684,36 @@
 y %7B%0A            
+   
+ 
 fnThread.join();
@@ -3705,32 +3705,36 @@
 nThread.join();%0A
+    
         %7D catch 
@@ -3768,24 +3768,28 @@
             
+   
+ 
 // ignore th
@@ -3791,16 +3791,30 @@
 re this%0A
+            %7D%0A
         
@@ -4656,39 +4656,133 @@
     
-return this.fnThread.isAlive();
+if (this.fnThread != null) %7B%0A            return this.fnThread.isAlive();%0A        %7D else %7B%0A            return false;%0A        %7D
 %0A   

Move the thread construction to start to avoid illegalstate (#266)