Call open before discover. This will avoid discover being called befo…

…re open (apache#7703)

Co-authored-by: Sanjeev Kulkarni <sanjeevk@splunk.com>

pulsar-io/batch/src/main/java/org/apache/pulsar/io/batch/BatchSourceExecutor.java

@@ -122,12 +122,12 @@ private void start() throws Exception {
     // This is the first thing to do to ensure that any tasks discovered during the discover
     // phase are not lost
     setupInstanceSubscription();
+    batchSource.open(this.config, this.sourceContext);
     if (sourceContext.getInstanceId() == 0) {
       discoveryTriggerer.init(batchSourceConfig.getDiscoveryTriggererConfig(),
                               this.sourceContext);
       discoveryTriggerer.start(this::triggerDiscover);
     }
-    batchSource.open(this.config, this.sourceContext);
   }
 
   private void triggerDiscover(String discoveredEvent) {