MINOR: Rename CoordinatorRuntime.activeTopicPartitions to activeCoordinators (#21010)

The coordinator runtime uses the term coordinator in all APIs so
`CoordinatorRuntime.activeTopicPartitions` feels a bit weird. This patch
renames it to `CoordinatorRuntime.activeCoordinators` to follow the
convention. It also makes the `state` field within the context volatile
as it is accesses without the log in that method.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Lianet Magrans <lmagrans@confluent.io>

Author: dajac

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

@@ -578,7 +578,7 @@ class CoordinatorContext {
         /**
          * The current state.
          */
-        CoordinatorState state;
+        volatile CoordinatorState state;
 
         /**
          * The current epoch of the coordinator. This represents
@@ -2611,14 +2611,9 @@ public void close() throws Exception {
     }
 
     /**
-     * Util method which returns all the topic partitions for which
-     * the state machine is in active state.
-     * <p>
-     * This could be useful if the caller does not have a specific
-     * target internal topic partition.
-     * @return List of {@link TopicPartition} whose coordinators are active
+     * @return List of {@link TopicPartition} whose coordinators are active.
      */
-    public List<TopicPartition> activeTopicPartitions() {
+    public List<TopicPartition> activeCoordinators() {
         if (coordinators == null || coordinators.isEmpty()) {
             return List.of();
         }

share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java

@@ -301,7 +301,7 @@ public void run() {
                     return;
                 }
                 List<CompletableFuture<Void>> futures = new ArrayList<>();
-                runtime.activeTopicPartitions().forEach(tp -> futures.add(performRecordPruning(tp)));
+                runtime.activeCoordinators().forEach(tp -> futures.add(performRecordPruning(tp)));
 
                 CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[]{}))
                     .whenComplete((res, exp) -> {

share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java

@@ -93,7 +93,7 @@ class ShareCoordinatorServiceTest {
     @SuppressWarnings("unchecked")
     private CoordinatorRuntime<ShareCoordinatorShard, CoordinatorRecord> mockRuntime() {
         CoordinatorRuntime<ShareCoordinatorShard, CoordinatorRecord> runtime = mock(CoordinatorRuntime.class);
-        when(runtime.activeTopicPartitions())
+        when(runtime.activeCoordinators())
             .thenReturn(List.of(new TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, 0)));
         return runtime;
     }
@@ -1533,7 +1533,7 @@ public void testRecordPruningTaskPeriodicityWithSomeFailures() throws Exception
         TopicPartition tp1 = new TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, 0);
         TopicPartition tp2 = new TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, 1);
 
-        when(runtime.activeTopicPartitions())
+        when(runtime.activeCoordinators())
             .thenReturn(List.of(tp1, tp2));
 
         when(writer.deleteRecords(