fix(share-group): move subscribe/startJob inside try so runLoop finally always guards appenderator

Author: Shekharrajak

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/ShareGroupIndexTaskRunner.java

@@ -275,9 +275,6 @@ TaskStatus runLoop(
       TaskToolbox toolbox
   ) throws Exception
   {
-    recordSupplier.subscribe(Collections.singleton(ioConfig.getTopic()));
-    driver.startJob(segmentId -> true);
-
     // Share groups manage delivery state on the broker; the Committer is a
     // no-op placeholder required by the appenderator driver contract.
     final Supplier<Committer> committerSupplier = () -> new Committer()
@@ -302,6 +299,8 @@ public void run()
     boolean appenderatorClosedNormally = false;
 
     try {
+      recordSupplier.subscribe(Collections.singleton(ioConfig.getTopic()));
+      driver.startJob(segmentId -> true);
       while (!task.isStopRequested()) {
 
         final List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity>> records;