solr/core/src/java/org/apache/solr/update/HdfsUpdateLog.java - lucene-solr - Git at Google

 /*
  * 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.solr.update;

 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.atomic.AtomicLong;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.lucene.util.BytesRef;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.SolrException.ErrorCode;
 import org.apache.solr.common.util.IOUtils;
 import org.apache.solr.core.PluginInfo;
 import org.apache.solr.core.SolrCore;
 import org.apache.solr.core.SolrInfoBean;
 import org.apache.solr.util.HdfsUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * @lucene.experimental
  * @deprecated since 8.6; see <a href="https://cwiki.apache.org/confluence/display/SOLR/Deprecations">Deprecations</a>
  */
 public class HdfsUpdateLog extends UpdateLog {

   private final Object fsLock = new Object();
   private FileSystem fs;
   private volatile Path tlogDir;
   private final String confDir;
   private Integer tlogDfsReplication;

   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
   private boolean debug = log.isDebugEnabled();

   // used internally by tests to track total count of failed tran log loads in init
   public static AtomicLong INIT_FAILED_LOGS_COUNT = new AtomicLong();

   public HdfsUpdateLog() {
     this(null);
   }

   public HdfsUpdateLog(String confDir) {
     this.confDir = confDir;
   }

   @Override
   public void init(PluginInfo info) {
     super.init(info);

     tlogDfsReplication = (Integer) info.initArgs.get( "tlogDfsReplication");
     if (tlogDfsReplication == null) tlogDfsReplication = 3;

     log.info("Initializing HdfsUpdateLog: tlogDfsReplication={}", tlogDfsReplication);
   }

   private Configuration getConf(Path path) {
     Configuration conf = new Configuration();
     if (confDir != null) {
       HdfsUtil.addHdfsResources(conf, confDir);
     }

     String fsScheme = path.toUri().getScheme();
     if(fsScheme != null) {
       conf.setBoolean("fs." + fsScheme + ".impl.disable.cache", true);
     }
     return conf;
   }

   @Override
   public void init(UpdateHandler uhandler, SolrCore core) {

     // ulogDir from CoreDescriptor overrides
     String ulogDir = core.getCoreDescriptor().getUlogDir();

     this.uhandler = uhandler;

     synchronized (fsLock) {
       // just like dataDir, we do not allow
       // moving the tlog dir on reload
       if (fs == null) {
         if (ulogDir != null) {
           dataDir = ulogDir;
         }
         if (dataDir == null || dataDir.length() == 0) {
           dataDir = core.getDataDir();
         }

         if (!core.getDirectoryFactory().isAbsolute(dataDir)) {
           try {
             dataDir = core.getDirectoryFactory().getDataHome(core.getCoreDescriptor());
           } catch (IOException e) {
             throw new SolrException(ErrorCode.SERVER_ERROR, e);
           }
         }

         try {
           Path dataDirPath = new Path(dataDir);
           fs = FileSystem.get(dataDirPath.toUri(), getConf(dataDirPath));
         } catch (IOException e) {
           throw new SolrException(ErrorCode.SERVER_ERROR, e);
         }
       } else {
         if (debug) {
           log.debug("UpdateHandler init: tlogDir={}, next id={}  this is a reopen or double init ... nothing else to do."
               , tlogDir, id);
         }
         versionInfo.reload();
         return;
       }
     }

     tlogDir = new Path(dataDir, TLOG_NAME);
     while (true) {
       try {
         if (!fs.exists(tlogDir)) {
           boolean success = fs.mkdirs(tlogDir);
           if (!success) {
             throw new RuntimeException("Could not create directory:" + tlogDir);
           }
         } else {
           fs.mkdirs(tlogDir); // To check for safe mode
         }
         break;
       } catch (RemoteException e) {
         if (e.getClassName().equals(
             "org.apache.hadoop.hdfs.server.namenode.SafeModeException")) {
           log.warn("The NameNode is in SafeMode - Solr will wait 5 seconds and try again.");
           try {
             Thread.sleep(5000);
           } catch (InterruptedException e1) {
             Thread.interrupted();
           }
           continue;
         }
         throw new RuntimeException(
             "Problem creating directory: " + tlogDir, e);
       } catch (IOException e) {
         throw new RuntimeException("Problem creating directory: " + tlogDir, e);
       }
     }

     String[] oldBufferTlog = getBufferLogList(fs, tlogDir);
     if (oldBufferTlog != null && oldBufferTlog.length != 0) {
       existOldBufferLog = true;
     }

     tlogFiles = getLogList(fs, tlogDir);
     id = getLastLogId() + 1; // add 1 since we will create a new log for the
                              // next update

     if (debug) {
       log.debug("UpdateHandler init: tlogDir={}, existing tlogs={}, next id={}"
           , tlogDir, Arrays.asList(tlogFiles), id);
     }

     TransactionLog oldLog = null;
     for (String oldLogName : tlogFiles) {
       Path f = new Path(tlogDir, oldLogName);
       try {
         oldLog = new HdfsTransactionLog(fs, f, null, true, tlogDfsReplication);
         addOldLog(oldLog, false); // don't remove old logs on startup since more
                                   // than one may be uncapped.
       } catch (Exception e) {
         INIT_FAILED_LOGS_COUNT.incrementAndGet();
         SolrException.log(log, "Failure to open existing log file (non fatal) "
             + f, e);
         try {
           fs.delete(f, false);
         } catch (IOException e1) {
           throw new RuntimeException(e1);
         }
       }
     }

     // Record first two logs (oldest first) at startup for potential tlog
     // recovery.
     // It's possible that at abnormal close both "tlog" and "prevTlog" were
     // uncapped.
     for (TransactionLog ll : logs) {
       if (newestLogsOnStartup.size() < 2) {
         newestLogsOnStartup.addFirst(ll);
       } else {
         // We're never going to modify old non-recovery logs - no need to hold their output open
         log.info("Closing output for old non-recovery log {}", ll);
         ll.closeOutput();
       }
     }

     try {
       versionInfo = new VersionInfo(this, numVersionBuckets);
     } catch (SolrException e) {
       log.error("Unable to use updateLog: ", e);
       throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
           "Unable to use updateLog: " + e.getMessage(), e);
     }

     // TODO: these startingVersions assume that we successfully recover from all
     // non-complete tlogs.
     try (RecentUpdates startingUpdates = getRecentUpdates()) {
       startingVersions = startingUpdates.getVersions(getNumRecordsToKeep());

       // populate recent deletes list (since we can't get that info from the
       // index)
       for (int i = startingUpdates.deleteList.size() - 1; i >= 0; i--) {
         DeleteUpdate du = startingUpdates.deleteList.get(i);
         oldDeletes.put(new BytesRef(du.id), new LogPtr(-1, du.version));
       }

       // populate recent deleteByQuery commands
       for (int i = startingUpdates.deleteByQueryList.size() - 1; i >= 0; i--) {
         Update update = startingUpdates.deleteByQueryList.get(i);
         @SuppressWarnings({"unchecked"})
         List<Object> dbq = (List<Object>) update.log.lookup(update.pointer);
         long version = (Long) dbq.get(1);
         String q = (String) dbq.get(2);
         trackDeleteByQuery(q, version);
       }

     }

     // initialize metrics
     core.getCoreMetricManager().registerMetricProducer(SolrInfoBean.Category.TLOG.toString(), this);
   }

   @Override
   public String getLogDir() {
     return tlogDir.toUri().toString();
   }

   public static String[] getBufferLogList(FileSystem fs, Path tlogDir) {
     final String prefix = BUFFER_TLOG_NAME+'.';
     assert fs != null;
     FileStatus[] fileStatuses;
     try {
       fileStatuses = fs.listStatus(tlogDir, path -> path.getName().startsWith(prefix));
     } catch (IOException e) {
       throw new SolrException(ErrorCode.SERVER_ERROR, "Failed on listing old buffer tlog", e);
     }

     String[] names = new String[fileStatuses.length];
     for (int i = 0; i < fileStatuses.length; i++) {
       names[i] = fileStatuses[i].getPath().getName();
     }
     return names;
   }

   public static String[] getLogList(FileSystem fs, Path tlogDir) {
     final String prefix = TLOG_NAME + '.';
     assert fs != null;
     FileStatus[] fileStatuses;
     try {
       fileStatuses = fs.listStatus(tlogDir, new PathFilter() {

         @Override
         public boolean accept(Path path) {
           return path.getName().startsWith(prefix);
         }
       });
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
     String[] names = new String[fileStatuses.length];
     for (int i = 0; i < fileStatuses.length; i++) {
       names[i] = fileStatuses[i].getPath().getName();
     }
     Arrays.sort(names);

     return names;
   }

   @Override
   public void close(boolean committed) {
     close(committed, false);
   }

   @Override
   public void close(boolean committed, boolean deleteOnClose) {
     try {
       super.close(committed, deleteOnClose);
     } finally {
       IOUtils.closeQuietly(fs);
     }
   }

   @Override
   protected void ensureBufferTlog() {
     if (bufferTlog != null) return;
     String newLogName = String.format(Locale.ROOT, LOG_FILENAME_PATTERN, BUFFER_TLOG_NAME, System.nanoTime());
     bufferTlog = new HdfsTransactionLog(fs, new Path(tlogDir, newLogName),
         globalStrings, tlogDfsReplication);
     bufferTlog.isBuffer = true;
   }

   @Override
   protected void deleteBufferLogs() {
     // Delete old buffer logs
     String[] oldBufferTlog = getBufferLogList(fs, tlogDir);
     if (oldBufferTlog != null && oldBufferTlog.length != 0) {
       for (String oldBufferLogName : oldBufferTlog) {
         Path f = new Path(tlogDir, oldBufferLogName);
         try {
           boolean s = fs.delete(f, false);
           if (!s) {
             log.error("Could not remove old buffer tlog file:{}", f);
           }
         } catch (IOException e) {
           // No need to bubble up this exception, because it won't cause any problems on recovering
           log.error("Could not remove old buffer tlog file:{}", f, e);
         }
       }
     }
   }

   @Override
   protected void ensureLog() {
     if (tlog == null) {
       String newLogName = String.format(Locale.ROOT, LOG_FILENAME_PATTERN,
           TLOG_NAME, id);
       HdfsTransactionLog ntlog = new HdfsTransactionLog(fs, new Path(tlogDir, newLogName),
           globalStrings, tlogDfsReplication);
       tlog = ntlog;
     }
   }

   /**
    * Clears the logs on the file system. Only call before init.
    *
    * @param core the SolrCore
    * @param ulogPluginInfo the init info for the UpdateHandler
    */
   @Override
   public void clearLog(SolrCore core, PluginInfo ulogPluginInfo) {
     if (ulogPluginInfo == null) return;
     Path tlogDir = new Path(getTlogDir(core, ulogPluginInfo));
     try {
       if (fs != null && fs.exists(tlogDir)) {
         String[] files = getLogList(tlogDir);
         for (String file : files) {
           Path f = new Path(tlogDir, file);
           boolean s = fs.delete(f, false);
           if (!s) {
             log.error("Could not remove tlog file:{}", f);
           }
         }
       }
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
   }

   @Override
   public void preSoftCommit(CommitUpdateCommand cmd) {
     debug = log.isDebugEnabled();
     super.preSoftCommit(cmd);
   }

   public String[] getLogList(Path tlogDir) throws FileNotFoundException, IOException {
     final String prefix = TLOG_NAME+'.';
     FileStatus[] files = fs.listStatus(tlogDir, new PathFilter() {

       @Override
       public boolean accept(Path name) {
         return name.getName().startsWith(prefix);
       }
     });
     List<String> fileList = new ArrayList<>(files.length);
     for (FileStatus file : files) {
       fileList.add(file.getPath().getName());
     }
     return fileList.toArray(new String[0]);
   }

   /**
    * Returns true if we were able to drop buffered updates and return to the
    * ACTIVE state
    */
   // public boolean dropBufferedUpdates() {
   // versionInfo.blockUpdates();
   // try {
   // if (state != State.BUFFERING) return false;
   //
   // if (log.isInfoEnabled()) {
   // log.info("Dropping buffered updates " + this);
   // }
   //
   // // since we blocked updates, this synchronization shouldn't strictly be
   // necessary.
   // synchronized (this) {
   // if (tlog != null) {
   // tlog.rollback(recoveryInfo.positionOfStart);
   // }
   // }
   //
   // state = State.ACTIVE;
   // operationFlags &= ~FLAG_GAP;
   // } catch (IOException e) {
   // SolrException.log(log,"Error attempting to roll back log", e);
   // return false;
   // }
   // finally {
   // versionInfo.unblockUpdates();
   // }
   // return true;
   // }

   public String toString() {
     return "HDFSUpdateLog{state=" + getState() + ", tlog=" + tlog + "}";
   }

 }
	/*
	* 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.solr.update;

	import java.io.FileNotFoundException;
	import java.io.IOException;
	import java.lang.invoke.MethodHandles;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.List;
	import java.util.Locale;
	import java.util.concurrent.atomic.AtomicLong;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.fs.PathFilter;
	import org.apache.hadoop.ipc.RemoteException;
	import org.apache.lucene.util.BytesRef;
	import org.apache.solr.common.SolrException;
	import org.apache.solr.common.SolrException.ErrorCode;
	import org.apache.solr.common.util.IOUtils;
	import org.apache.solr.core.PluginInfo;
	import org.apache.solr.core.SolrCore;
	import org.apache.solr.core.SolrInfoBean;
	import org.apache.solr.util.HdfsUtil;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* @lucene.experimental
	* @deprecated since 8.6; see <a href="https://cwiki.apache.org/confluence/display/SOLR/Deprecations">Deprecations</a>
	*/
	public class HdfsUpdateLog extends UpdateLog {

	private final Object fsLock = new Object();
	private FileSystem fs;
	private volatile Path tlogDir;
	private final String confDir;
	private Integer tlogDfsReplication;

	private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
	private boolean debug = log.isDebugEnabled();

	// used internally by tests to track total count of failed tran log loads in init
	public static AtomicLong INIT_FAILED_LOGS_COUNT = new AtomicLong();

	public HdfsUpdateLog() {
	this(null);
	}

	public HdfsUpdateLog(String confDir) {
	this.confDir = confDir;
	}

	@Override
	public void init(PluginInfo info) {
	super.init(info);

	tlogDfsReplication = (Integer) info.initArgs.get( "tlogDfsReplication");
	if (tlogDfsReplication == null) tlogDfsReplication = 3;

	log.info("Initializing HdfsUpdateLog: tlogDfsReplication={}", tlogDfsReplication);
	}

	private Configuration getConf(Path path) {
	Configuration conf = new Configuration();
	if (confDir != null) {
	HdfsUtil.addHdfsResources(conf, confDir);
	}

	String fsScheme = path.toUri().getScheme();
	if(fsScheme != null) {
	conf.setBoolean("fs." + fsScheme + ".impl.disable.cache", true);
	}
	return conf;
	}

	@Override
	public void init(UpdateHandler uhandler, SolrCore core) {

	// ulogDir from CoreDescriptor overrides
	String ulogDir = core.getCoreDescriptor().getUlogDir();

	this.uhandler = uhandler;

	synchronized (fsLock) {
	// just like dataDir, we do not allow
	// moving the tlog dir on reload
	if (fs == null) {
	if (ulogDir != null) {
	dataDir = ulogDir;
	}
	if (dataDir == null \|\| dataDir.length() == 0) {
	dataDir = core.getDataDir();
	}

	if (!core.getDirectoryFactory().isAbsolute(dataDir)) {
	try {
	dataDir = core.getDirectoryFactory().getDataHome(core.getCoreDescriptor());
	} catch (IOException e) {
	throw new SolrException(ErrorCode.SERVER_ERROR, e);
	}
	}

	try {
	Path dataDirPath = new Path(dataDir);
	fs = FileSystem.get(dataDirPath.toUri(), getConf(dataDirPath));
	} catch (IOException e) {
	throw new SolrException(ErrorCode.SERVER_ERROR, e);
	}
	} else {
	if (debug) {
	log.debug("UpdateHandler init: tlogDir={}, next id={} this is a reopen or double init ... nothing else to do."
	, tlogDir, id);
	}
	versionInfo.reload();
	return;
	}
	}

	tlogDir = new Path(dataDir, TLOG_NAME);
	while (true) {
	try {
	if (!fs.exists(tlogDir)) {
	boolean success = fs.mkdirs(tlogDir);
	if (!success) {
	throw new RuntimeException("Could not create directory:" + tlogDir);
	}
	} else {
	fs.mkdirs(tlogDir); // To check for safe mode
	}
	break;
	} catch (RemoteException e) {
	if (e.getClassName().equals(
	"org.apache.hadoop.hdfs.server.namenode.SafeModeException")) {
	log.warn("The NameNode is in SafeMode - Solr will wait 5 seconds and try again.");
	try {
	Thread.sleep(5000);
	} catch (InterruptedException e1) {
	Thread.interrupted();
	}
	continue;
	}
	throw new RuntimeException(
	"Problem creating directory: " + tlogDir, e);
	} catch (IOException e) {
	throw new RuntimeException("Problem creating directory: " + tlogDir, e);
	}
	}

	String[] oldBufferTlog = getBufferLogList(fs, tlogDir);
	if (oldBufferTlog != null && oldBufferTlog.length != 0) {
	existOldBufferLog = true;
	}

	tlogFiles = getLogList(fs, tlogDir);
	id = getLastLogId() + 1; // add 1 since we will create a new log for the
	// next update

	if (debug) {
	log.debug("UpdateHandler init: tlogDir={}, existing tlogs={}, next id={}"
	, tlogDir, Arrays.asList(tlogFiles), id);
	}

	TransactionLog oldLog = null;
	for (String oldLogName : tlogFiles) {
	Path f = new Path(tlogDir, oldLogName);
	try {
	oldLog = new HdfsTransactionLog(fs, f, null, true, tlogDfsReplication);
	addOldLog(oldLog, false); // don't remove old logs on startup since more
	// than one may be uncapped.
	} catch (Exception e) {
	INIT_FAILED_LOGS_COUNT.incrementAndGet();
	SolrException.log(log, "Failure to open existing log file (non fatal) "
	+ f, e);
	try {
	fs.delete(f, false);
	} catch (IOException e1) {
	throw new RuntimeException(e1);
	}
	}
	}

	// Record first two logs (oldest first) at startup for potential tlog
	// recovery.
	// It's possible that at abnormal close both "tlog" and "prevTlog" were
	// uncapped.
	for (TransactionLog ll : logs) {
	if (newestLogsOnStartup.size() < 2) {
	newestLogsOnStartup.addFirst(ll);
	} else {
	// We're never going to modify old non-recovery logs - no need to hold their output open
	log.info("Closing output for old non-recovery log {}", ll);
	ll.closeOutput();
	}
	}

	try {
	versionInfo = new VersionInfo(this, numVersionBuckets);
	} catch (SolrException e) {
	log.error("Unable to use updateLog: ", e);
	throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
	"Unable to use updateLog: " + e.getMessage(), e);
	}

	// TODO: these startingVersions assume that we successfully recover from all
	// non-complete tlogs.
	try (RecentUpdates startingUpdates = getRecentUpdates()) {
	startingVersions = startingUpdates.getVersions(getNumRecordsToKeep());

	// populate recent deletes list (since we can't get that info from the
	// index)
	for (int i = startingUpdates.deleteList.size() - 1; i >= 0; i--) {
	DeleteUpdate du = startingUpdates.deleteList.get(i);
	oldDeletes.put(new BytesRef(du.id), new LogPtr(-1, du.version));
	}

	// populate recent deleteByQuery commands
	for (int i = startingUpdates.deleteByQueryList.size() - 1; i >= 0; i--) {
	Update update = startingUpdates.deleteByQueryList.get(i);
	@SuppressWarnings({"unchecked"})
	List<Object> dbq = (List<Object>) update.log.lookup(update.pointer);
	long version = (Long) dbq.get(1);
	String q = (String) dbq.get(2);
	trackDeleteByQuery(q, version);
	}

	}

	// initialize metrics
	core.getCoreMetricManager().registerMetricProducer(SolrInfoBean.Category.TLOG.toString(), this);
	}

	@Override
	public String getLogDir() {
	return tlogDir.toUri().toString();
	}

	public static String[] getBufferLogList(FileSystem fs, Path tlogDir) {
	final String prefix = BUFFER_TLOG_NAME+'.';
	assert fs != null;
	FileStatus[] fileStatuses;
	try {
	fileStatuses = fs.listStatus(tlogDir, path -> path.getName().startsWith(prefix));
	} catch (IOException e) {
	throw new SolrException(ErrorCode.SERVER_ERROR, "Failed on listing old buffer tlog", e);
	}

	String[] names = new String[fileStatuses.length];
	for (int i = 0; i < fileStatuses.length; i++) {
	names[i] = fileStatuses[i].getPath().getName();
	}
	return names;
	}

	public static String[] getLogList(FileSystem fs, Path tlogDir) {
	final String prefix = TLOG_NAME + '.';
	assert fs != null;
	FileStatus[] fileStatuses;
	try {
	fileStatuses = fs.listStatus(tlogDir, new PathFilter() {

	@Override
	public boolean accept(Path path) {
	return path.getName().startsWith(prefix);
	}
	});
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	String[] names = new String[fileStatuses.length];
	for (int i = 0; i < fileStatuses.length; i++) {
	names[i] = fileStatuses[i].getPath().getName();
	}
	Arrays.sort(names);

	return names;
	}

	@Override
	public void close(boolean committed) {
	close(committed, false);
	}

	@Override
	public void close(boolean committed, boolean deleteOnClose) {
	try {
	super.close(committed, deleteOnClose);
	} finally {
	IOUtils.closeQuietly(fs);
	}
	}

	@Override
	protected void ensureBufferTlog() {
	if (bufferTlog != null) return;
	String newLogName = String.format(Locale.ROOT, LOG_FILENAME_PATTERN, BUFFER_TLOG_NAME, System.nanoTime());
	bufferTlog = new HdfsTransactionLog(fs, new Path(tlogDir, newLogName),
	globalStrings, tlogDfsReplication);
	bufferTlog.isBuffer = true;
	}

	@Override
	protected void deleteBufferLogs() {
	// Delete old buffer logs
	String[] oldBufferTlog = getBufferLogList(fs, tlogDir);
	if (oldBufferTlog != null && oldBufferTlog.length != 0) {
	for (String oldBufferLogName : oldBufferTlog) {
	Path f = new Path(tlogDir, oldBufferLogName);
	try {
	boolean s = fs.delete(f, false);
	if (!s) {
	log.error("Could not remove old buffer tlog file:{}", f);
	}
	} catch (IOException e) {
	// No need to bubble up this exception, because it won't cause any problems on recovering
	log.error("Could not remove old buffer tlog file:{}", f, e);
	}
	}
	}
	}

	@Override
	protected void ensureLog() {
	if (tlog == null) {
	String newLogName = String.format(Locale.ROOT, LOG_FILENAME_PATTERN,
	TLOG_NAME, id);
	HdfsTransactionLog ntlog = new HdfsTransactionLog(fs, new Path(tlogDir, newLogName),
	globalStrings, tlogDfsReplication);
	tlog = ntlog;
	}
	}

	/**
	* Clears the logs on the file system. Only call before init.
	*
	* @param core the SolrCore
	* @param ulogPluginInfo the init info for the UpdateHandler
	*/
	@Override
	public void clearLog(SolrCore core, PluginInfo ulogPluginInfo) {
	if (ulogPluginInfo == null) return;
	Path tlogDir = new Path(getTlogDir(core, ulogPluginInfo));
	try {
	if (fs != null && fs.exists(tlogDir)) {
	String[] files = getLogList(tlogDir);
	for (String file : files) {
	Path f = new Path(tlogDir, file);
	boolean s = fs.delete(f, false);
	if (!s) {
	log.error("Could not remove tlog file:{}", f);
	}
	}
	}
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	}

	@Override
	public void preSoftCommit(CommitUpdateCommand cmd) {
	debug = log.isDebugEnabled();
	super.preSoftCommit(cmd);
	}

	public String[] getLogList(Path tlogDir) throws FileNotFoundException, IOException {
	final String prefix = TLOG_NAME+'.';
	FileStatus[] files = fs.listStatus(tlogDir, new PathFilter() {

	@Override
	public boolean accept(Path name) {
	return name.getName().startsWith(prefix);
	}
	});
	List<String> fileList = new ArrayList<>(files.length);
	for (FileStatus file : files) {
	fileList.add(file.getPath().getName());
	}
	return fileList.toArray(new String[0]);
	}

	/**
	* Returns true if we were able to drop buffered updates and return to the
	* ACTIVE state
	*/
	// public boolean dropBufferedUpdates() {
	// versionInfo.blockUpdates();
	// try {
	// if (state != State.BUFFERING) return false;
	//
	// if (log.isInfoEnabled()) {
	// log.info("Dropping buffered updates " + this);
	// }
	//
	// // since we blocked updates, this synchronization shouldn't strictly be
	// necessary.
	// synchronized (this) {
	// if (tlog != null) {
	// tlog.rollback(recoveryInfo.positionOfStart);
	// }
	// }
	//
	// state = State.ACTIVE;
	// operationFlags &= ~FLAG_GAP;
	// } catch (IOException e) {
	// SolrException.log(log,"Error attempting to roll back log", e);
	// return false;
	// }
	// finally {
	// versionInfo.unblockUpdates();
	// }
	// return true;
	// }

	public String toString() {
	return "HDFSUpdateLog{state=" + getState() + ", tlog=" + tlog + "}";
	}

	}