Xinfra monitor code formatting changes and Proper Indentations for MultiClusterTopicManagementService class

Xinfra monitor code formatting changes and Proper Indentations for MultiClusterTopicManagementService class Signed-off-by: Andrew Choi <li_andchoi@microsoft.com>

Author: Andrew Choi

src/main/java/com/linkedin/xinfra/monitor/services/MultiClusterTopicManagementService.java

@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import kafka.admin.AdminUtils;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -34,7 +35,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
-import kafka.admin.AdminUtils;
 import kafka.admin.BrokerMetadata;
 import kafka.server.ConfigType;
 import kafka.zk.KafkaZkClient;
@@ -88,17 +88,18 @@ public class MultiClusterTopicManagementService implements Service {
  private final long _preferredLeaderElectionIntervalMs;
  private final ScheduledExecutorService _executor;
 
-
  @SuppressWarnings("unchecked")
  public MultiClusterTopicManagementService(Map<String, Object> props, String serviceName) throws Exception {
  _serviceName = serviceName;
  MultiClusterTopicManagementServiceConfig config = new MultiClusterTopicManagementServiceConfig(props);
  String topic = config.getString(CommonServiceConfig.TOPIC_CONFIG);
- Map<String, Map> propsByCluster = props.containsKey(MultiClusterTopicManagementServiceConfig.PROPS_PER_CLUSTER_CONFIG)
- ? (Map) props.get(MultiClusterTopicManagementServiceConfig.PROPS_PER_CLUSTER_CONFIG) : new HashMap<>();
+ Map<String, Map> propsByCluster =
+ props.containsKey(MultiClusterTopicManagementServiceConfig.PROPS_PER_CLUSTER_CONFIG) ? (Map) props.get(
+ MultiClusterTopicManagementServiceConfig.PROPS_PER_CLUSTER_CONFIG) : new HashMap<>();
  _topicManagementByCluster = initializeTopicManagementHelper(propsByCluster, topic);
  _scheduleIntervalMs = config.getInt(MultiClusterTopicManagementServiceConfig.REBALANCE_INTERVAL_MS_CONFIG);
- _preferredLeaderElectionIntervalMs = config.getLong(MultiClusterTopicManagementServiceConfig.PREFERRED_LEADER_ELECTION_CHECK_INTERVAL_MS_CONFIG);
+ _preferredLeaderElectionIntervalMs =
+ config.getLong(MultiClusterTopicManagementServiceConfig.PREFERRED_LEADER_ELECTION_CHECK_INTERVAL_MS_CONFIG);
  _executor = Executors.newSingleThreadScheduledExecutor(
  r -> new Thread(r, _serviceName + "-multi-cluster-topic-management-service"));
  _topicPartitionResult.complete(null);
@@ -108,14 +109,16 @@ public CompletableFuture<Void> topicPartitionResult() {
  return _topicPartitionResult;
  }
 
- private Map<String, TopicManagementHelper> initializeTopicManagementHelper(Map<String, Map> propsByCluster, String topic) throws Exception {
+ private Map<String, TopicManagementHelper> initializeTopicManagementHelper(Map<String, Map> propsByCluster,
+ String topic) throws Exception {
  Map<String, TopicManagementHelper> topicManagementByCluster = new HashMap<>();
- for (Map.Entry<String, Map> entry: propsByCluster.entrySet()) {
+ for (Map.Entry<String, Map> entry : propsByCluster.entrySet()) {
  String clusterName = entry.getKey();
  Map serviceProps = entry.getValue();
- if (serviceProps.containsKey(MultiClusterTopicManagementServiceConfig.TOPIC_CONFIG))
- throw new ConfigException("The raw per-cluster config for MultiClusterTopicManagementService must not contain " +
- MultiClusterTopicManagementServiceConfig.TOPIC_CONFIG);
+ if (serviceProps.containsKey(MultiClusterTopicManagementServiceConfig.TOPIC_CONFIG)) {
+ throw new ConfigException("The raw per-cluster config for MultiClusterTopicManagementService must not contain "
+ + MultiClusterTopicManagementServiceConfig.TOPIC_CONFIG);
+ }
  serviceProps.put(MultiClusterTopicManagementServiceConfig.TOPIC_CONFIG, topic);
  topicManagementByCluster.put(clusterName, new TopicManagementHelper(serviceProps));
  }
@@ -129,8 +132,8 @@ public synchronized void start() {
  _executor.scheduleWithFixedDelay(tmRunnable, 0, _scheduleIntervalMs, TimeUnit.MILLISECONDS);
 
  Runnable pleRunnable = new PreferredLeaderElectionRunnable();
- _executor.scheduleWithFixedDelay(pleRunnable, _preferredLeaderElectionIntervalMs, _preferredLeaderElectionIntervalMs,
- TimeUnit.MILLISECONDS);
+ _executor.scheduleWithFixedDelay(pleRunnable, _preferredLeaderElectionIntervalMs,
+ _preferredLeaderElectionIntervalMs, TimeUnit.MILLISECONDS);
  LOGGER.info("{}/MultiClusterTopicManagementService started.", _serviceName);
  }
  }
@@ -153,13 +156,12 @@ public void awaitShutdown(long timeout, TimeUnit unit) {
  try {
  _executor.awaitTermination(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
  } catch (InterruptedException e) {
- LOGGER.info("Thread interrupted when waiting for {}/MultiClusterTopicManagementService to shutdown", _serviceName);
+ LOGGER.info("Thread interrupted when waiting for {}/MultiClusterTopicManagementService to shutdown",
+ _serviceName);
  }
  LOGGER.info("{}/MultiClusterTopicManagementService shutdown completed", _serviceName);
  }
 
-
-
  private class TopicManagementRunnable implements Runnable {
 
  @Override
@@ -190,7 +192,8 @@ public void run() {
  try {
  helper.maybeReassignPartitionAndElectLeader();
  } catch (KafkaException e) {
- LOGGER.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName, e);
+ LOGGER.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName,
+ e);
  }
  }
  } catch (Throwable t) {
@@ -216,7 +219,8 @@ public void run() {
  try {
  helper.maybeElectLeader();
  } catch (IOException | KafkaException e) {
- LOGGER.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName, e);
+ LOGGER.warn(_serviceName + "/MultiClusterTopicManagementService will retry later in cluster " + clusterName,
+ e);
  }
  }
  } catch (Throwable t) {
@@ -246,7 +250,6 @@ static class TopicManagementHelper {
  String _topic;
  TopicFactory _topicFactory;
 
-
  @SuppressWarnings("unchecked")
  TopicManagementHelper(Map<String, Object> props) throws Exception {
  TopicManagementServiceConfig config = new TopicManagementServiceConfig(props);
@@ -263,13 +266,17 @@ static class TopicManagementHelper {
  _bootstrapServers = adminClientConfig.getList(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG);
  _topicProperties = new Properties();
  if (props.containsKey(TopicManagementServiceConfig.TOPIC_PROPS_CONFIG)) {
- for (Map.Entry<String, Object> entry: ((Map<String, Object>) props.get(TopicManagementServiceConfig.TOPIC_PROPS_CONFIG)).entrySet())
+ for (Map.Entry<String, Object> entry : ((Map<String, Object>) props.get(
+ TopicManagementServiceConfig.TOPIC_PROPS_CONFIG)).entrySet()) {
  _topicProperties.put(entry.getKey(), entry.getValue().toString());
+ }
  }
 
- Map topicFactoryConfig = props.containsKey(TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) ?
- (Map) props.get(TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) : new HashMap();
- _topicFactory = (TopicFactory) Class.forName(topicFactoryClassName).getConstructor(Map.class).newInstance(topicFactoryConfig);
+ Map topicFactoryConfig =
+ props.containsKey(TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) ? (Map) props.get(
+ TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) : new HashMap();
+ _topicFactory =
+ (TopicFactory) Class.forName(topicFactoryClassName).getConstructor(Map.class).newInstance(topicFactoryConfig);
 
  _adminClient = constructAdminClient(props);
  LOGGER.info("{} configs: {}", _adminClient.getClass().getSimpleName(), props);
@@ -304,29 +311,27 @@ void maybeAddPartitions(int minPartitionNum) throws ExecutionException, Interrup
 
  int partitionNum = partitions.size();
  if (partitionNum < minPartitionNum) {
- LOGGER.info("{} will increase partition of the topic {} in the cluster from {}"
- + " to {}.", this.getClass().toString(), _topic, partitionNum, minPartitionNum);
+ LOGGER.info("{} will increase partition of the topic {} in the cluster from {}" + " to {}.",
+ this.getClass().toString(), _topic, partitionNum, minPartitionNum);
  Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_zkConnect);
  Set<BrokerMetadata> brokers = new HashSet<>();
  for (Node broker : _adminClient.describeCluster().nodes().get()) {
- BrokerMetadata brokerMetadata = new BrokerMetadata(
- broker.id(), null
- );
+ BrokerMetadata brokerMetadata = new BrokerMetadata(broker.id(), null);
  brokers.add(brokerMetadata);
  }
 
  if (!blackListedBrokers.isEmpty()) {
  brokers.removeIf(broker -> blackListedBrokers.contains(broker.id()));
  }
 
- List<List<Integer>> newPartitionAssignments = newPartitionAssignments(minPartitionNum, partitionNum, brokers, _replicationFactor);
+ List<List<Integer>> newPartitionAssignments =
+ newPartitionAssignments(minPartitionNum, partitionNum, brokers, _replicationFactor);
 
  NewPartitions newPartitions = NewPartitions.increaseTo(minPartitionNum, newPartitionAssignments);
 
  Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
  newPartitionsMap.put(_topic, newPartitions);
  CreatePartitionsResult createPartitionsResult = _adminClient.createPartitions(newPartitionsMap);
-
  }
  }
 
@@ -402,7 +407,6 @@ int numPartitions() throws InterruptedException, ExecutionException {
  return _adminClient.describeTopics(Collections.singleton(_topic)).values().get(_topic).get().partitions().size();
  }
 
-
  private Set<Node> getAvailableBrokers() throws ExecutionException, InterruptedException {
  Set<Node> brokers = new HashSet<>(_adminClient.describeCluster().nodes().get());
  Set<Integer> blackListedBrokers = _topicFactory.getBlackListedBrokers(_zkConnect);
@@ -412,11 +416,11 @@ private Set<Node> getAvailableBrokers() throws ExecutionException, InterruptedEx
 
  void maybeReassignPartitionAndElectLeader() throws ExecutionException, InterruptedException, TimeoutException {
  try (KafkaZkClient zkClient = KafkaZkClient.apply(_zkConnect, JaasUtils.isZkSecurityEnabled(),
- Utils.ZK_SESSION_TIMEOUT_MS, Utils.ZK_CONNECTION_TIMEOUT_MS,
- Integer.MAX_VALUE, Time.SYSTEM, METRIC_GROUP_NAME, "SessionExpireListener", null)) {
+ Utils.ZK_SESSION_TIMEOUT_MS, Utils.ZK_CONNECTION_TIMEOUT_MS, Integer.MAX_VALUE, Time.SYSTEM,
+ METRIC_GROUP_NAME, "SessionExpireListener", null)) {
 
- List<TopicPartitionInfo> partitionInfoList = _adminClient
- .describeTopics(Collections.singleton(_topic)).all().get().get(_topic).partitions();
+ List<TopicPartitionInfo> partitionInfoList =
+ _adminClient.describeTopics(Collections.singleton(_topic)).all().get().get(_topic).partitions();
  Collection<Node> brokers = this.getAvailableBrokers();
  boolean partitionReassigned = false;
  if (partitionInfoList.size() == 0) {
@@ -426,27 +430,31 @@ void maybeReassignPartitionAndElectLeader() throws ExecutionException, Interrupt
  int currentReplicationFactor = getReplicationFactor(partitionInfoList);
  int expectedReplicationFactor = Math.max(currentReplicationFactor, _replicationFactor);
 
- if (_replicationFactor < currentReplicationFactor)
+ if (_replicationFactor < currentReplicationFactor) {
  LOGGER.debug(
  "Configured replication factor {} is smaller than the current replication factor {} of the topic {} in cluster.",
  _replicationFactor, currentReplicationFactor, _topic);
+ }
 
- if (expectedReplicationFactor > currentReplicationFactor
- && Utils.ongoingPartitionReassignments(_adminClient).isEmpty()) {
+ if (expectedReplicationFactor > currentReplicationFactor && Utils.ongoingPartitionReassignments(_adminClient)
+ .isEmpty()) {
  LOGGER.info(
  "MultiClusterTopicManagementService will increase the replication factor of the topic {} in cluster"
  + "from {} to {}", _topic, currentReplicationFactor, expectedReplicationFactor);
  reassignPartitions(_adminClient, brokers, _topic, partitionInfoList.size(), expectedReplicationFactor);
+
  partitionReassigned = true;
  }
 
  // Update the properties of the monitor topic if any config is different from the user-specified config
  Properties currentProperties = zkClient.getEntityConfigs(ConfigType.Topic(), _topic);
  Properties expectedProperties = new Properties();
- for (Object key : currentProperties.keySet())
+ for (Object key : currentProperties.keySet()) {
  expectedProperties.put(key, currentProperties.get(key));
- for (Object key : _topicProperties.keySet())
+ }
+ for (Object key : _topicProperties.keySet()) {
  expectedProperties.put(key, _topicProperties.get(key));
+ }
 
  if (!currentProperties.equals(expectedProperties)) {
  LOGGER.info("MultiClusterTopicManagementService will overwrite properties of the topic {} "
@@ -458,16 +466,14 @@ void maybeReassignPartitionAndElectLeader() throws ExecutionException, Interrupt
  && Utils.ongoingPartitionReassignments(_adminClient).isEmpty()) {
  LOGGER.info("{} will reassign partitions of the topic {} in cluster.", this.getClass().toString(), _topic);
  reassignPartitions(_adminClient, brokers, _topic, partitionInfoList.size(), expectedReplicationFactor);
+
  partitionReassigned = true;
  }
 
- if (partitionInfoList.size() >= brokers.size() &&
- someBrokerNotElectedLeader(partitionInfoList, brokers)) {
+ if (partitionInfoList.size() >= brokers.size() && someBrokerNotElectedLeader(partitionInfoList, brokers)) {
  if (!partitionReassigned || Utils.ongoingPartitionReassignments(_adminClient).isEmpty()) {
- LOGGER.info(
- "MultiClusterTopicManagementService will trigger preferred leader election for the topic {} in "
- + "cluster.", _topic
- );
+ LOGGER.info("MultiClusterTopicManagementService will trigger preferred leader election for the topic {} in "
+ + "cluster.", _topic);
  triggerPreferredLeaderElection(partitionInfoList, _topic);
  _preferredLeaderElectionRequested = false;
  } else {
@@ -482,14 +488,14 @@ void maybeElectLeader() throws Exception {
  return;
  }
 
- try (KafkaZkClient zkClient = KafkaZkClient.apply(_zkConnect, JaasUtils.isZkSecurityEnabled(), Utils.ZK_SESSION_TIMEOUT_MS,
- Utils.ZK_CONNECTION_TIMEOUT_MS, Integer.MAX_VALUE, Time.SYSTEM, METRIC_GROUP_NAME, "SessionExpireListener", null)) {
+ try (KafkaZkClient zkClient = KafkaZkClient.apply(_zkConnect, JaasUtils.isZkSecurityEnabled(),
+ Utils.ZK_SESSION_TIMEOUT_MS, Utils.ZK_CONNECTION_TIMEOUT_MS, Integer.MAX_VALUE, Time.SYSTEM,
+ METRIC_GROUP_NAME, "SessionExpireListener", null)) {
  if (Utils.ongoingPartitionReassignments(_adminClient).isEmpty()) {
- List<TopicPartitionInfo> partitionInfoList = _adminClient
- .describeTopics(Collections.singleton(_topic)).all().get().get(_topic).partitions();
- LOGGER.info(
- "MultiClusterTopicManagementService will trigger requested preferred leader election for the"
- + " topic {} in cluster.", _topic);
+ List<TopicPartitionInfo> partitionInfoList =
+ _adminClient.describeTopics(Collections.singleton(_topic)).all().get().get(_topic).partitions();
+ LOGGER.info("MultiClusterTopicManagementService will trigger requested preferred leader election for the"
+ + " topic {} in cluster.", _topic);
  triggerPreferredLeaderElection(partitionInfoList, _topic);
  _preferredLeaderElectionRequested = false;
  }
@@ -503,7 +509,8 @@ private void triggerPreferredLeaderElection(List<TopicPartitionInfo> partitionIn
  }
  ElectPreferredLeadersResult electPreferredLeadersResult = _adminClient.electPreferredLeaders(partitions);
 
- LOGGER.info("{}: triggerPreferredLeaderElection - {}", this.getClass().toString(), electPreferredLeadersResult.all());
+ LOGGER.info("{}: triggerPreferredLeaderElection - {}", this.getClass().toString(),
+ electPreferredLeadersResult.all());
  }
 
  private static void reassignPartitions(AdminClient adminClient, Collection<Node> brokers, String topic,
@@ -541,11 +548,13 @@ private static void reassignPartitions(AdminClient adminClient, Collection<Node>
  NewPartitionReassignment newPartitionReassignment = new NewPartitionReassignment(targetReplicas);
  reassignments.put(topicPartitionSeqEntry.getKey(), Optional.of(newPartitionReassignment));
  }
+
  AlterPartitionReassignmentsResult alterPartitionReassignmentsResult =
  adminClient.alterPartitionReassignments(reassignments);
  try {
  alterPartitionReassignmentsResult.all().get();
  } catch (InterruptedException | ExecutionException e) {
+
  LOGGER.error("An exception occurred while altering the partition reassignments for {}", topic, e);
  }
  }
@@ -567,21 +576,25 @@ static int getReplicationFactor(List<TopicPartitionInfo> partitionInfoList) {
 
  static boolean someBrokerNotPreferredLeader(List<TopicPartitionInfo> partitionInfoList, Collection<Node> brokers) {
  Set<Integer> brokersNotPreferredLeader = new HashSet<>(brokers.size());
- for (Node broker: brokers)
+ for (Node broker : brokers) {
  brokersNotPreferredLeader.add(broker.id());
- for (TopicPartitionInfo partitionInfo : partitionInfoList)
+ }
+ for (TopicPartitionInfo partitionInfo : partitionInfoList) {
  brokersNotPreferredLeader.remove(partitionInfo.replicas().get(0).id());
+ }
 
  return !brokersNotPreferredLeader.isEmpty();
  }
 
  static boolean someBrokerNotElectedLeader(List<TopicPartitionInfo> partitionInfoList, Collection<Node> brokers) {
  Set<Integer> brokersNotElectedLeader = new HashSet<>(brokers.size());
- for (Node broker: brokers)
+ for (Node broker : brokers) {
  brokersNotElectedLeader.add(broker.id());
+ }
  for (TopicPartitionInfo partitionInfo : partitionInfoList) {
- if (partitionInfo.leader() != null)
+ if (partitionInfo.leader() != null) {
  brokersNotElectedLeader.remove(partitionInfo.leader().id());
+ }
  }
  return !brokersNotElectedLeader.isEmpty();
  }
@@ -634,11 +647,16 @@ static boolean someBrokerNotElectedLeader(List<TopicPartitionInfo> partitionInfo
  * {"topic":"kmf-topic","partition":0,"replicas":[2,0]}]}
  * </pre>
  */
- private static String formatAsNewReassignmentJson(String topic, scala.collection.Map<Object, Seq<Object>> partitionsToReassign) {
+ private static String formatAsNewReassignmentJson(String topic,
+ scala.collection.Map<Object, Seq<Object>> partitionsToReassign) {
  StringBuilder builder = new StringBuilder();
  builder.append("{\"version\":1,\"partitions\":[\n");
  for (int partition = 0; partition < partitionsToReassign.size(); partition++) {
- builder.append(" {\"topic\":\"").append(topic).append("\",\"partition\":").append(partition).append(",\"replicas\":[");
+ builder.append(" {\"topic\":\"")
+ .append(topic)
+ .append("\",\"partition\":")
+ .append(partition)
+ .append(",\"replicas\":[");
  Seq<Object> replicas = partitionsToReassign.apply(partition);
  for (int replicaIndex = 0; replicaIndex < replicas.size(); replicaIndex++) {
  Object replica = replicas.apply(replicaIndex);
@@ -651,6 +669,5 @@ private static String formatAsNewReassignmentJson(String topic, scala.collection
  builder.append("]}");
  return builder.toString();
  }
-
  }
 }