Add more information about function in getstatus (apache#205)

Author: srkukarni

pulsar-functions/proto/src/main/proto/InstanceCommunication.proto

@@ -27,6 +27,7 @@ option java_outer_classname = "InstanceCommunication";
 message FunctionStatus {
     bool running = 1;
     string failureException = 2;
+    int64 numRestarts = 11;
     int64 numProcessed = 3;
     int64 numSuccessfullyProcessed = 4;
     int64 numTimeouts = 5;

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

@@ -70,6 +70,7 @@ public static Spawner createSpawner(FunctionConfig fnConfig,
     private MetricsSink metricsSink;
     private int metricsCollectionInterval;
     private Timer metricsCollectionTimer;
+    private int numRestarts;
 
     private Spawner(LimitsConfig limitsConfig,
                     AssignmentInfo assignmentInfo,
@@ -83,12 +84,14 @@ private Spawner(LimitsConfig limitsConfig,
         this.codeFile = codeFile;
         this.metricsSink = metricsSink;
         this.metricsCollectionInterval = metricsCollectionInterval;
+        this.numRestarts = 0;
     }
 
     public void start() throws Exception {
         log.info("Spawner starting function {}", this.assignmentInfo.getFunctionConfig().getName());
         functionContainer = functionContainerFactory.createContainer(createJavaInstanceConfig(), codeFile);
         functionContainer.start();
+        numRestarts++;
         if (metricsSink != null) {
             log.info("Scheduling Metrics Collection every " + metricsCollectionInterval + " secs for " + FunctionConfigUtils.getFullyQualifiedName(assignmentInfo.getFunctionConfig()));
             metricsCollectionTimer = new Timer();
@@ -107,6 +110,7 @@ public void run() {
                         log.error("Function Container is dead with exception", functionContainer.getDeathException());
                         log.error("Restarting...");
                         functionContainer.start();
+                        numRestarts++;
                     }
                 }
             }, metricsCollectionInterval * 1000, metricsCollectionInterval * 1000);
@@ -120,7 +124,14 @@ public void join() throws Exception {
     }
 
     public CompletableFuture<FunctionStatus> getFunctionStatus() {
-        return functionContainer.getFunctionStatus();
+        return functionContainer.getFunctionStatus().thenApply(f -> {
+           FunctionStatus.Builder builder = FunctionStatus.newBuilder();
+           builder.mergeFrom(f).setNumRestarts(numRestarts);
+           if (functionContainer.getDeathException() != null) {
+               builder.setFailureException(functionContainer.getDeathException().getMessage());
+           }
+           return builder.build();
+        });
     }
 
     @Override