[SPARK-50568][SS][TESTS] Fix KafkaMicroBatchSourceSuite to cover KafkaOffsetReaderConsumer

### What changes were proposed in this pull request?

This was missed when the default value of configuration is changed via the following.
- #38306

KafkaMicroBatchSourceSuite consists of set of different suites, where KafkaMicroBatchSourceSuiteBase based suite is defined. There are 4 implementations of this abstract class for now:
  1. `KafkaMicroBatchV1SourceSuite` - V1 source that supposes to use `KafkaOffsetReaderConsumer` as `KafkaOffsetReader`.
  2. `KafkaMicroBatchV2SourceSuite` - V2 source that supposes to use `KafkaOffsetReaderConsumer` as `KafkaOffsetReader`.
  3. `KafkaMicroBatchV1SourceWithAdminSuite` - V1 source that uses `KafkaOffsetReaderAdmin` as `KafkaOffsetReader`.
  4. `KafkaMicroBatchV2SourceWithAdminSuite` - V2 source that uses `KafkaOffsetReaderAdmin` as `KafkaOffsetReader`.
But `KafkaMicroBatchV1SourceSuite` and `KafkaMicroBatchV2SourceSuite` are still running based on `KafkaOffsetReaderAdmin`, as `USE_DEPRECATED_KAFKA_OFFSET_FETCHING` is `false` be default. By switching it to `true` in `beforeAll`, we make sure that corresponding `KafkaOffsetReader` is in use.

### Why are the changes needed?

To improve unit tests coverage for `KafkaOffsetReaderConsumer`

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Unit Tests.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #49164 from ostronaut/hotifx/KafkaMicroBatchSourceSuite-cover-KafkaOffsetReaderConsumer.

Authored-by: Dima <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>

Author: ostronaut

connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala

@@ -1591,22 +1591,7 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase with
   }
 }
 
-
-class KafkaMicroBatchV1SourceWithAdminSuite extends KafkaMicroBatchV1SourceSuite {
-  override def beforeAll(): Unit = {
-    super.beforeAll()
-    spark.conf.set(SQLConf.USE_DEPRECATED_KAFKA_OFFSET_FETCHING.key, "false")
-  }
-}
-
-class KafkaMicroBatchV2SourceWithAdminSuite extends KafkaMicroBatchV2SourceSuite {
-  override def beforeAll(): Unit = {
-    super.beforeAll()
-    spark.conf.set(SQLConf.USE_DEPRECATED_KAFKA_OFFSET_FETCHING.key, "false")
-  }
-}
-
-class KafkaMicroBatchV1SourceSuite extends KafkaMicroBatchSourceSuiteBase {
+abstract class KafkaMicroBatchV1SourceSuite extends KafkaMicroBatchSourceSuiteBase {
   override def beforeAll(): Unit = {
     super.beforeAll()
     spark.conf.set(
@@ -1637,7 +1622,7 @@ class KafkaMicroBatchV1SourceSuite extends KafkaMicroBatchSourceSuiteBase {
   }
 }
 
-class KafkaMicroBatchV2SourceSuite extends KafkaMicroBatchSourceSuiteBase {
+abstract class KafkaMicroBatchV2SourceSuite extends KafkaMicroBatchSourceSuiteBase {
 
   test("V2 Source is used by default") {
     val topic = newTopic()
@@ -1870,6 +1855,35 @@ class KafkaMicroBatchV2SourceSuite extends KafkaMicroBatchSourceSuiteBase {
   }
 }
 
+class KafkaMicroBatchV1SourceWithAdminSuite extends KafkaMicroBatchV1SourceSuite {
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    spark.conf.set(SQLConf.USE_DEPRECATED_KAFKA_OFFSET_FETCHING.key, "false")
+  }
+}
+
+class KafkaMicroBatchV1SourceWithConsumerSuite extends KafkaMicroBatchV1SourceSuite {
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    spark.conf.set(SQLConf.USE_DEPRECATED_KAFKA_OFFSET_FETCHING.key, "true")
+  }
+}
+
+class KafkaMicroBatchV2SourceWithAdminSuite extends KafkaMicroBatchV2SourceSuite {
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    spark.conf.set(SQLConf.USE_DEPRECATED_KAFKA_OFFSET_FETCHING.key, "false")
+  }
+}
+
+class KafkaMicroBatchV2SourceWithConsumerSuite extends KafkaMicroBatchV2SourceSuite {
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    spark.conf.set(SQLConf.USE_DEPRECATED_KAFKA_OFFSET_FETCHING.key, "true")
+  }
+}
+
+
 abstract class KafkaSourceSuiteBase extends KafkaSourceTest {
 
   import testImplicits._