<commit_before>/*
 * 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.accumulo.server.tabletserver.log;

import static org.apache.accumulo.server.logger.LogEvents.COMPACTION_FINISH;
import static org.apache.accumulo.server.logger.LogEvents.COMPACTION_START;
import static org.apache.accumulo.server.logger.LogEvents.DEFINE_TABLET;
import static org.apache.accumulo.server.logger.LogEvents.MANY_MUTATIONS;
import static org.apache.accumulo.server.logger.LogEvents.MUTATION;
import static org.apache.accumulo.server.logger.LogEvents.OPEN;

import java.io.IOException;
import java.util.List;
import java.util.Set;

import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.logger.LogFileKey;
import org.apache.accumulo.server.logger.LogFileValue;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * Extract Mutations for a tablet from a set of logs that have been sorted by operation and tablet.
 * 
 */
public class SortedLogRecovery {
  private static final Logger log = Logger.getLogger(SortedLogRecovery.class);
  
  static class EmptyMapFileException extends Exception {
    private static final long serialVersionUID = 1L;

    public EmptyMapFileException() { super(); }
  }

  static class UnusedException extends Exception {
    private static final long serialVersionUID = 1L;

    public UnusedException() { super(); }
  }

  private VolumeManager fs;

  public SortedLogRecovery(VolumeManager fs) {
    this.fs = fs;
  }
  
  private enum Status {
    INITIAL, LOOKING_FOR_FINISH, COMPLETE
  };
  
  private static class LastStartToFinish {
    long lastStart = -1;
    long seq = -1;
    long lastFinish = -1;
    Status compactionStatus = Status.INITIAL;
    String tserverSession = "";
    
    private void update(long newFinish) {
      this.seq = this.lastStart;
      if (newFinish != -1)
        lastFinish = newFinish;
    }
    
    private void update(int newStartFile, long newStart) {
      this.lastStart = newStart;
    }
    
    private void update(String newSession) {
      this.lastStart = -1;
      this.lastFinish = -1;
      this.compactionStatus = Status.INITIAL;
      this.tserverSession = newSession;
    }
  }
  
  public void recover(KeyExtent extent, List<Path> recoveryLogs, Set<String> tabletFiles, MutationReceiver mr) throws IOException {
    int[] tids = new int[recoveryLogs.size()];
    LastStartToFinish lastStartToFinish = new LastStartToFinish();
    for (int i = 0; i < recoveryLogs.size(); i++) {
      Path logfile = recoveryLogs.get(i);
      log.info("Looking at mutations from " + logfile + " for " + extent);
      MultiReader reader = new MultiReader(fs, logfile);
      try {
        try {
          tids[i] = findLastStartToFinish(reader, i, extent, tabletFiles, lastStartToFinish);
        } catch (EmptyMapFileException ex) {
          log.info("Ignoring empty map file " + logfile);
          tids[i] = -1;
        } catch (UnusedException ex) {
          log.info("Ignoring log file " + logfile + " appears to be unused by " + extent);
          tids[i] = -1;
        }
      } finally {
        try {
          reader.close();
        } catch (IOException ex) {
          log.warn("Ignoring error closing file");
        }
      }
      
    }
    
    if (lastStartToFinish.compactionStatus == Status.LOOKING_FOR_FINISH)
      throw new RuntimeException("COMPACTION_FINISH (without preceding COMPACTION_START) not followed by successful minor compaction");
    
    for (int i = 0; i < recoveryLogs.size(); i++) {
      Path logfile = recoveryLogs.get(i);
      MultiReader reader = new MultiReader(fs, logfile);
      try {
        playbackMutations(reader, tids[i], lastStartToFinish, mr);
      } finally {
        try {
          reader.close();
        } catch (IOException ex) {
          log.warn("Ignoring error closing file");
        }
      }
      log.info("Recovery complete for " + extent + " using " + logfile);
    }
  }
  
  int findLastStartToFinish(MultiReader reader, int fileno, KeyExtent extent, Set<String> tabletFiles, LastStartToFinish lastStartToFinish) throws IOException, EmptyMapFileException, UnusedException {
    // Scan for tableId for this extent (should always be in the log)
    LogFileKey key = new LogFileKey();
    LogFileValue value = new LogFileValue();
    int tid = -1;
    if (!reader.next(key, value))
      throw new EmptyMapFileException();
    if (key.event != OPEN)
      throw new RuntimeException("First log entry value is not OPEN");
    
    if (key.tserverSession.compareTo(lastStartToFinish.tserverSession) != 0) {
      if (lastStartToFinish.compactionStatus == Status.LOOKING_FOR_FINISH)
        throw new RuntimeException("COMPACTION_FINISH (without preceding COMPACTION_START) is not followed by a successful minor compaction.");
      lastStartToFinish.update(key.tserverSession);
    }
    
    LogFileKey defineKey = null;
    
    // find the maximum tablet id... because a tablet may leave a tserver and then come back, in which case it would have a different tablet id
    // for the maximum tablet id, find the minimum sequence #... may be ok to find the max seq, but just want to make the code behave like it used to
    while (reader.next(key, value)) {
      // LogReader.printEntry(entry);
      if (key.event != DEFINE_TABLET)
        break;
      if (key.tablet.equals(extent)) {
        if (tid != key.tid) {
          tid = key.tid;
          defineKey = key;
          key = new LogFileKey();
        }
      }
    }
    if (tid < 0) {
      throw new UnusedException();
    }
    
    log.debug("Found tid, seq " + tid + " " + defineKey.seq);
    
    // Scan start/stop events for this tablet
    key = defineKey;
    key.event = COMPACTION_START;
    reader.seek(key);
    while (reader.next(key, value)) {
      // LogFileEntry.printEntry(entry);
      if (key.tid != tid)
        break;
      if (key.event == COMPACTION_START) {
        if (lastStartToFinish.compactionStatus == Status.INITIAL)
          lastStartToFinish.compactionStatus = Status.COMPLETE;
        if (key.seq <= lastStartToFinish.lastStart)
          throw new RuntimeException("Sequence numbers are not increasing for start/stop events.");
        lastStartToFinish.update(fileno, key.seq);
        
        // Tablet server finished the minor compaction, but didn't remove the entry from the METADATA table.
        log.error("filename in compaction start " + key.filename);
        if (tabletFiles.contains(key.filename))
          lastStartToFinish.update(-1);
      } else if (key.event == COMPACTION_FINISH) {
        if (key.seq <= lastStartToFinish.lastStart)
          throw new RuntimeException("Sequence numbers are not increasing for start/stop events.");
        if (lastStartToFinish.compactionStatus == Status.INITIAL)
          lastStartToFinish.compactionStatus = Status.LOOKING_FOR_FINISH;
        else if (lastStartToFinish.lastFinish > lastStartToFinish.lastStart)
          throw new RuntimeException("COMPACTION_FINISH does not have preceding COMPACTION_START event.");
        else
          lastStartToFinish.compactionStatus = Status.COMPLETE;<commit_msg>ACCUMULO-1811 demote message, and make the message more informative<commit_after>/*
 * 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.accumulo.server.tabletserver.log;

import static org.apache.accumulo.server.logger.LogEvents.COMPACTION_FINISH;
import static org.apache.accumulo.server.logger.LogEvents.COMPACTION_START;
import static org.apache.accumulo.server.logger.LogEvents.DEFINE_TABLET;
import static org.apache.accumulo.server.logger.LogEvents.MANY_MUTATIONS;
import static org.apache.accumulo.server.logger.LogEvents.MUTATION;
import static org.apache.accumulo.server.logger.LogEvents.OPEN;

import java.io.IOException;
import java.util.List;
import java.util.Set;

import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.logger.LogFileKey;
import org.apache.accumulo.server.logger.LogFileValue;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * Extract Mutations for a tablet from a set of logs that have been sorted by operation and tablet.
 * 
 */
public class SortedLogRecovery {
  private static final Logger log = Logger.getLogger(SortedLogRecovery.class);
  
  static class EmptyMapFileException extends Exception {
    private static final long serialVersionUID = 1L;

    public EmptyMapFileException() { super(); }
  }

  static class UnusedException extends Exception {
    private static final long serialVersionUID = 1L;

    public UnusedException() { super(); }
  }

  private VolumeManager fs;

  public SortedLogRecovery(VolumeManager fs) {
    this.fs = fs;
  }
  
  private enum Status {
    INITIAL, LOOKING_FOR_FINISH, COMPLETE
  };
  
  private static class LastStartToFinish {
    long lastStart = -1;
    long seq = -1;
    long lastFinish = -1;
    Status compactionStatus = Status.INITIAL;
    String tserverSession = "";
    
    private void update(long newFinish) {
      this.seq = this.lastStart;
      if (newFinish != -1)
        lastFinish = newFinish;
    }
    
    private void update(int newStartFile, long newStart) {
      this.lastStart = newStart;
    }
    
    private void update(String newSession) {
      this.lastStart = -1;
      this.lastFinish = -1;
      this.compactionStatus = Status.INITIAL;
      this.tserverSession = newSession;
    }
  }
  
  public void recover(KeyExtent extent, List<Path> recoveryLogs, Set<String> tabletFiles, MutationReceiver mr) throws IOException {
    int[] tids = new int[recoveryLogs.size()];
    LastStartToFinish lastStartToFinish = new LastStartToFinish();
    for (int i = 0; i < recoveryLogs.size(); i++) {
      Path logfile = recoveryLogs.get(i);
      log.info("Looking at mutations from " + logfile + " for " + extent);
      MultiReader reader = new MultiReader(fs, logfile);
      try {
        try {
          tids[i] = findLastStartToFinish(reader, i, extent, tabletFiles, lastStartToFinish);
        } catch (EmptyMapFileException ex) {
          log.info("Ignoring empty map file " + logfile);
          tids[i] = -1;
        } catch (UnusedException ex) {
          log.info("Ignoring log file " + logfile + " appears to be unused by " + extent);
          tids[i] = -1;
        }
      } finally {
        try {
          reader.close();
        } catch (IOException ex) {
          log.warn("Ignoring error closing file");
        }
      }
      
    }
    
    if (lastStartToFinish.compactionStatus == Status.LOOKING_FOR_FINISH)
      throw new RuntimeException("COMPACTION_FINISH (without preceding COMPACTION_START) not followed by successful minor compaction");
    
    for (int i = 0; i < recoveryLogs.size(); i++) {
      Path logfile = recoveryLogs.get(i);
      MultiReader reader = new MultiReader(fs, logfile);
      try {
        playbackMutations(reader, tids[i], lastStartToFinish, mr);
      } finally {
        try {
          reader.close();
        } catch (IOException ex) {
          log.warn("Ignoring error closing file");
        }
      }
      log.info("Recovery complete for " + extent + " using " + logfile);
    }
  }
  
  int findLastStartToFinish(MultiReader reader, int fileno, KeyExtent extent, Set<String> tabletFiles, LastStartToFinish lastStartToFinish) throws IOException, EmptyMapFileException, UnusedException {
    // Scan for tableId for this extent (should always be in the log)
    LogFileKey key = new LogFileKey();
    LogFileValue value = new LogFileValue();
    int tid = -1;
    if (!reader.next(key, value))
      throw new EmptyMapFileException();
    if (key.event != OPEN)
      throw new RuntimeException("First log entry value is not OPEN");
    
    if (key.tserverSession.compareTo(lastStartToFinish.tserverSession) != 0) {
      if (lastStartToFinish.compactionStatus == Status.LOOKING_FOR_FINISH)
        throw new RuntimeException("COMPACTION_FINISH (without preceding COMPACTION_START) is not followed by a successful minor compaction.");
      lastStartToFinish.update(key.tserverSession);
    }
    
    LogFileKey defineKey = null;
    
    // find the maximum tablet id... because a tablet may leave a tserver and then come back, in which case it would have a different tablet id
    // for the maximum tablet id, find the minimum sequence #... may be ok to find the max seq, but just want to make the code behave like it used to
    while (reader.next(key, value)) {
      // LogReader.printEntry(entry);
      if (key.event != DEFINE_TABLET)
        break;
      if (key.tablet.equals(extent)) {
        if (tid != key.tid) {
          tid = key.tid;
          defineKey = key;
          key = new LogFileKey();
        }
      }
    }
    if (tid < 0) {
      throw new UnusedException();
    }
    
    log.debug("Found tid, seq " + tid + " " + defineKey.seq);
    
    // Scan start/stop events for this tablet
    key = defineKey;
    key.event = COMPACTION_START;
    reader.seek(key);
    while (reader.next(key, value)) {
      // LogFileEntry.printEntry(entry);
      if (key.tid != tid)
        break;
      if (key.event == COMPACTION_START) {
        if (lastStartToFinish.compactionStatus == Status.INITIAL)
          lastStartToFinish.compactionStatus = Status.COMPLETE;
        if (key.seq <= lastStartToFinish.lastStart)
          throw new RuntimeException("Sequence numbers are not increasing for start/stop events.");
        lastStartToFinish.update(fileno, key.seq);
        
        // Tablet server finished the minor compaction, but didn't remove the entry from the METADATA table.
        log.debug("minor compaction into " + key.filename + " finished, but was still in the METADATA");
        if (tabletFiles.contains(key.filename))
          lastStartToFinish.update(-1);
      } else if (key.event == COMPACTION_FINISH) {
        if (key.seq <= lastStartToFinish.lastStart)
          throw new RuntimeException("Sequence numbers are not increasing for start/stop events.");
        if (lastStartToFinish.compactionStatus == Status.INITIAL)
          lastStartToFinish.compactionStatus = Status.LOOKING_FOR_FINISH;
        else if (lastStartToFinish.lastFinish > lastStartToFinish.lastStart)
          throw new RuntimeException("COMPACTION_FINISH does not have preceding COMPACTION_START event.");
        else
          lastStartToFinish.compactionStatus = Status.COMPLETE;
{'fraction_non_alphanumeric': 0.09325717124053148, 'fraction_numerical': 0.0028560784800695394, 'mean_word_length': 3.024737631184408, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5030434', 'n_tokens_mistral': 4685, 'n_tokens_neox': 4554, 'n_words': 1365}