Improve naming of threads used in batch source

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/batch/BatchSourceExecutor.java

@@ -69,14 +69,19 @@ public class BatchSourceExecutor<T> implements Source<T> {
   private String intermediateTopicName;
   private volatile Exception currentError = null;
   private volatile boolean isRunning = false;
-  private ExecutorService discoveryThread = Executors.newSingleThreadExecutor(new DefaultThreadFactory("batch-source-discovery"));
+  private ExecutorService discoveryThread;
 
   @Override
   public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
     this.config = config;
     this.sourceContext = sourceContext;
     this.intermediateTopicName = SourceConfigUtils.computeBatchSourceIntermediateTopicName(sourceContext.getTenant(),
       sourceContext.getNamespace(), sourceContext.getSourceName()).toString();
+    this.discoveryThread = Executors.newSingleThreadExecutor(
+      new DefaultThreadFactory(
+        String.format("%s-batch-source-discovery",
+          FunctionCommon.getFullyQualifiedName(
+            sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()))));
     this.getBatchSourceConfigs(config);
     this.initializeBatchSource();
     this.start();

pulsar-io/batch-discovery-triggerers/src/main/java/org/apache/pulsar/io/batchdiscovery/CronTriggerer.java

@@ -46,12 +46,16 @@ public void init(Map<String, Object> config, SourceContext sourceContext) {
     } else {
       throw new IllegalArgumentException("Cron Trigger is not provided with Cron String");
     }
+    scheduler = new ThreadPoolTaskScheduler();
+    scheduler.setThreadNamePrefix(String.format("%s/%s/%s-cron-triggerer-",
+      sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()));
+
     log.info("Initialized CronTrigger with expression: {}", cronExpression);
   }
 
   @Override
   public void start(Consumer<String> trigger) {
-    scheduler = new ThreadPoolTaskScheduler();
+
     scheduler.initialize();
     scheduler.schedule(() -> trigger.accept("CRON"), new CronTrigger(cronExpression));
   }