content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.cassandra.tools;
  
  import java.io.IOException;
  import java.io.PrintStream;
  import java.text.SimpleDateFormat;
  import java.util.List;
  import javax.management.ListenerNotFoundException;
  import javax.management.remote.JMXConnector;
  
  import org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus;
  import org.apache.cassandra.service.StorageServiceMBean;
  import org.apache.cassandra.utils.Closeable;
  import org.apache.cassandra.utils.concurrent.Condition;
  import org.apache.cassandra.utils.progress.ProgressEvent;
  import org.apache.cassandra.utils.progress.ProgressEventType;
  import org.apache.cassandra.utils.progress.jmx.JMXNotificationProgressListener;
  
  import static java.util.concurrent.TimeUnit.SECONDS;
  import static org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus.FAILED;
  import static org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus.valueOf;
  import static org.apache.cassandra.tools.NodeProbe.JMX_NOTIFICATION_POLL_INTERVAL_SECONDS;
  import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
  import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
  import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition;
  import static org.apache.cassandra.utils.progress.ProgressEventType.COMPLETE;
  import static org.apache.cassandra.utils.progress.ProgressEventType.ERROR;
  import static org.apache.cassandra.utils.progress.ProgressEventType.PROGRESS;
  
  public class RepairRunner extends JMXNotificationProgressListener implements Closeable
  {
      public static abstract class RepairCmd
      {
          private final String keyspace;
  
          public RepairCmd(String keyspace)
          {
              this.keyspace = keyspace;
          }
  
          public abstract Integer start();
      }
      private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS");
  
      private final PrintStream out;
      private final JMXConnector jmxc;
      private final StorageServiceMBean ssProxy;
      private final Condition condition = newOneTimeCondition();
  
      private final RepairCmd repairCmd;
      private Integer cmd;
      private volatile Exception error;
  
      public RepairRunner(PrintStream out, JMXConnector jmxc, StorageServiceMBean ssProxy, RepairCmd repairCmd)
      {
          this.out = out;
          this.jmxc = jmxc;
          this.ssProxy = ssProxy;
          this.repairCmd = repairCmd;
      }
  
      public void start()
      {
          if (jmxc != null)
              jmxc.addConnectionNotificationListener(this, null, null);
          ssProxy.addNotificationListener(this, null, null);
          this.cmd = repairCmd.start();
      }
  
      @Override
      public void close()
      {
          try
          {
              ssProxy.removeNotificationListener(this);
          }
          catch (ListenerNotFoundException e)
          {
              // noop - there may be double removes with error handling
          }
          if (jmxc != null)
          {
              try
              {
                  jmxc.removeConnectionNotificationListener(this);
              }
              catch (ListenerNotFoundException e)
              {
                  // noop - there may be double removes with error handling
              }
          }
      }
  
      public void run() throws Exception
      {
          run(false);
      }
  
      public void run(boolean block) throws Exception
      {
          if (cmd == null)
              return;
  
          if (cmd <= 0)
          {
              // repairAsync can only return 0 for replication factor 1.
              String message = String.format("Replication factor is 1. No repair is needed for keyspace '%s'", repairCmd.keyspace);
              printMessage(message);
          }
          else
          {
              if (block)
              {
                  condition.await();
              }
              else
              {
                  while (!condition.await(JMX_NOTIFICATION_POLL_INTERVAL_SECONDS, SECONDS))
                  {
                      queryForCompletedRepair(String.format("After waiting for poll interval of %s seconds",
                                                            JMX_NOTIFICATION_POLL_INTERVAL_SECONDS));
                  }
              }
              Exception error = this.error;
              if (error == null)
              {
                  // notifications are lossy so its possible to see complete and not error; request latest state
                  // from the server
                  queryForCompletedRepair("condition satisfied");
                  error = this.error;
              }
              if (error != null)
              {
                  throw error;
              }
          }
      }
  
      @Override
      public boolean isInterestedIn(String tag)
      {
          return tag.equals("repair:" + cmd);
      }
  
      @Override
      public void handleNotificationLost(long timestamp, String message)
      {
          if (cmd > 0)
          {
              // Check to see if the lost notification was a completion message
              queryForCompletedRepair("After receiving lost notification");
          }
      }
  
      @Override
      public void handleConnectionClosed(long timestamp, String message)
      {
          handleConnectionFailed(timestamp, message);
      }
  
      @Override
      public void handleConnectionFailed(long timestamp, String message)
      {
          error = new IOException(String.format("[%s] JMX connection closed. You should check server log for repair status of keyspace %s"
                                                + "(Subsequent keyspaces are not going to be repaired).",
                                                format.format(timestamp), repairCmd.keyspace));
          condition.signalAll();
      }
  
      @Override
      public void progress(String tag, ProgressEvent event)
      {
          ProgressEventType type = event.getType();
          String message = event.getMessage();
          if (type == PROGRESS)
          {
              message = message + " (progress: " + (int) event.getProgressPercentage() + "%)";
          }
          printMessage(message);
          if (type == ERROR)
          {
              error = new RuntimeException(String.format("Repair job has failed with the error message: %s. " +
                                                         "Check the logs on the repair participants for further details",
                                                         message));
          }
          if (type == COMPLETE)
          {
              condition.signalAll();
          }
      }
  
  
      private void queryForCompletedRepair(String triggeringCondition)
      {
          List<String> status = ssProxy.getParentRepairStatus(cmd);
          String queriedString = "queried for parent session status and";
          if (status == null)
          {
              String message = String.format("%s %s couldn't find repair status for cmd: %s", triggeringCondition,
                                             queriedString, cmd);
              printMessage(message);
          }
          else
          {
              ParentRepairStatus parentRepairStatus = valueOf(status.get(0));
              List<String> messages = status.subList(1, status.size());
              switch (parentRepairStatus)
              {
                  case COMPLETED:
                  case FAILED:
                      printMessage(String.format("%s %s discovered repair %s.",
                                                triggeringCondition,
                                                queriedString, toLowerCaseLocalized(parentRepairStatus.name())));
                      if (parentRepairStatus == FAILED)
                      {
                          error = new IOException(messages.get(0));
                      }
                      printMessages(messages);
                      condition.signalAll();
                      break;
                  case IN_PROGRESS:
                      break;
                  default:
                      printMessage(String.format("WARNING Encountered unexpected RepairRunnable.ParentRepairStatus: %s", parentRepairStatus));
                      printMessages(messages);
                      break;
              }
          }
      }
  
      private void printMessages(List<String> messages)
      {
          for (String message : messages)
          {
              printMessage(message);
          }
      }
  
      private void printMessage(String message)
      {
          out.println(String.format("[%s] %s", this.format.format(currentTimeMillis()), message));
      }
  }
  