Misc Function fixes:- (apache#3907)

1) Collect input topics from the function details spec
2) Catch all errors during source/sink close since its user code

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java

@@ -76,8 +76,6 @@ class ContextImpl implements Context, SinkContext, SourceContext {
     private Map<String, Producer<?>> publishProducers;
     private ProducerBuilderImpl<?> producerBuilder;
 
-    private final List<String> inputTopics;
-
     private final TopicSchema topicSchema;
 
     private final SecretsProvider secretsProvider;
@@ -102,13 +100,12 @@ class ContextImpl implements Context, SinkContext, SourceContext {
     }
     private final Utils.ComponentType componentType;
 
-    public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client, List<String> inputTopics,
+    public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client,
                        SecretsProvider secretsProvider, CollectorRegistry collectorRegistry, String[] metricsLabels,
                        Utils.ComponentType componentType, ComponentStatsManager statsManager) {
         this.config = config;
         this.logger = logger;
         this.publishProducers = new HashMap<>();
-        this.inputTopics = inputTopics;
         this.topicSchema = new TopicSchema(client);
         this.statsManager = statsManager;
 
@@ -169,7 +166,7 @@ public Record<?> getCurrentRecord() {
 
     @Override
     public Collection<String> getInputTopics() {
-        return inputTopics;
+        return config.getFunctionDetails().getSource().getInputSpecsMap().keySet();
     }
 
     @Override

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

@@ -207,13 +207,9 @@ JavaInstance setupJavaInstance(ContextImpl contextImpl) throws Exception {
     }
 
     ContextImpl setupContext() {
-        List<String> inputTopics = null;
-        if (source instanceof PulsarSource) {
-            inputTopics = ((PulsarSource<?>) source).getInputTopics();
-        }
         Logger instanceLog = LoggerFactory.getLogger(
                 "function-" + instanceConfig.getFunctionDetails().getName());
-        return new ContextImpl(instanceConfig, instanceLog, client, inputTopics, secretsProvider,
+        return new ContextImpl(instanceConfig, instanceLog, client, secretsProvider,
                 collectorRegistry, metricsLabels, this.componentType, this.stats);
     }
 
@@ -470,7 +466,7 @@ public void close() {
         if (source != null) {
             try {
                 source.close();
-            } catch (Exception e) {
+            } catch (Throwable e) {
                 log.error("Failed to close source {}", instanceConfig.getFunctionDetails().getSource().getClassName(), e);
 
             }
@@ -479,7 +475,7 @@ public void close() {
         if (sink != null) {
             try {
                 sink.close();
-            } catch (Exception e) {
+            } catch (Throwable e) {
                 log.error("Failed to close sink {}", instanceConfig.getFunctionDetails().getSource().getClassName(), e);
             }
         }

pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java

@@ -77,7 +77,6 @@ public void setup() {
             config,
             logger,
             client,
-            new ArrayList<>(),
             new EnvironmentBasedSecretsProvider(), new CollectorRegistry(), new String[0],
                 Utils.ComponentType.FUNCTION, null);
     }