branch-2.0.4-alpha/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestEditLogJournalFailures.java - hadoop - 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.hadoop.hdfs.server.namenode;

 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.spy;

 import java.io.File;
 import java.io.IOException;

 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.server.namenode.JournalSet.JournalAndStream;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.ExitUtil.ExitException;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;

 public class TestEditLogJournalFailures {

   private int editsPerformed = 0;
   private MiniDFSCluster cluster;
   private FileSystem fs;

   /**
    * Create the mini cluster for testing and sub in a custom runtime so that
    * edit log journal failures don't actually cause the JVM to exit.
    */
   @Before
   public void setUpMiniCluster() throws IOException {
     setUpMiniCluster(new HdfsConfiguration(), true);
   }

   public void setUpMiniCluster(Configuration conf, boolean manageNameDfsDirs)
       throws IOException {
     cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0)
         .manageNameDfsDirs(manageNameDfsDirs).checkExitOnShutdown(false).build();
     cluster.waitActive();
     fs = cluster.getFileSystem();
   }

   @After
   public void shutDownMiniCluster() throws IOException {
     if (fs != null)
       fs.close();
     if (cluster != null) {
       try {
         cluster.shutdown();
       } catch (ExitException ee) {
         // Ignore ExitExceptions as the tests may result in the
         // NameNode doing an immediate shutdown.
       }
     }
   }

   @Test
   public void testSingleFailedEditsDirOnFlush() throws IOException {
     assertTrue(doAnEdit());
     // Invalidate one edits journal.
     invalidateEditsDirAtIndex(0, true, false);
     // The NN has not terminated (no ExitException thrown)
     assertTrue(doAnEdit());
     // A single journal failure should not result in a call to terminate
     assertFalse(cluster.getNameNode().isInSafeMode());
   }

   @Test
   public void testAllEditsDirsFailOnFlush() throws IOException {
     assertTrue(doAnEdit());
     // Invalidate both edits journals.
     invalidateEditsDirAtIndex(0, true, false);
     invalidateEditsDirAtIndex(1, true, false);
     // The NN has not terminated (no ExitException thrown)
     try {
       doAnEdit();
       fail("The previous edit could not be synced to any persistent storage, "
           + "should have halted the NN");
     } catch (RemoteException re) {
       assertTrue(re.getClassName().contains("ExitException"));
       GenericTestUtils.assertExceptionContains(
           "Could not sync enough journals to persistent storage. " +
           "Unsynced transactions: 1", re);
     }
   }

   @Test
   public void testAllEditsDirFailOnWrite() throws IOException {
     assertTrue(doAnEdit());
     // Invalidate both edits journals.
     invalidateEditsDirAtIndex(0, true, true);
     invalidateEditsDirAtIndex(1, true, true);
     // The NN has not terminated (no ExitException thrown)
     try {
       doAnEdit();
       fail("The previous edit could not be synced to any persistent storage, "
           + " should have halted the NN");
     } catch (RemoteException re) {
       assertTrue(re.getClassName().contains("ExitException"));
       GenericTestUtils.assertExceptionContains(
           "Could not sync enough journals to persistent storage due to " +
           "No journals available to flush. " +
           "Unsynced transactions: 1", re);
     }
   }

   @Test
   public void testSingleFailedEditsDirOnSetReadyToFlush() throws IOException {
     assertTrue(doAnEdit());
     // Invalidate one edits journal.
     invalidateEditsDirAtIndex(0, false, false);
     // The NN has not terminated (no ExitException thrown)
     assertTrue(doAnEdit());
     // A single journal failure should not result in a call to terminate
     assertFalse(cluster.getNameNode().isInSafeMode());
   }

   @Test
   public void testSingleRequiredFailedEditsDirOnSetReadyToFlush()
       throws IOException {
     // Set one of the edits dirs to be required.
     String[] editsDirs = cluster.getConfiguration(0).getTrimmedStrings(
         DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY);
     shutDownMiniCluster();
     Configuration conf = new HdfsConfiguration();
     conf.set(DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_REQUIRED_KEY, editsDirs[0]);
     conf.setInt(DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_KEY, 0);
     conf.setInt(DFSConfigKeys.DFS_NAMENODE_CHECKED_VOLUMES_MINIMUM_KEY, 0);
     setUpMiniCluster(conf, true);

     assertTrue(doAnEdit());
     // Invalidated the one required edits journal.
     invalidateEditsDirAtIndex(0, false, false);
     JournalAndStream nonRequiredJas = getJournalAndStream(1);
     EditLogFileOutputStream nonRequiredSpy =
       spyOnStream(nonRequiredJas);

     // The NN has not terminated (no ExitException thrown)

     // ..and that the other stream is active.
     assertTrue(nonRequiredJas.isActive());

     try {
       doAnEdit();
       fail("A single failure of a required journal should have halted the NN");
     } catch (RemoteException re) {
       assertTrue(re.getClassName().contains("ExitException"));
       GenericTestUtils.assertExceptionContains(
           "setReadyToFlush failed for required journal", re);
     }

     // Since the required directory failed setReadyToFlush, and that
     // directory was listed prior to the non-required directory,
     // we should not call setReadyToFlush on the non-required
     // directory. Regression test for HDFS-2874.
     Mockito.verify(nonRequiredSpy, Mockito.never()).setReadyToFlush();
     assertFalse(nonRequiredJas.isActive());
   }

   @Test
   public void testMultipleRedundantFailedEditsDirOnSetReadyToFlush()
       throws IOException {
     // Set up 4 name/edits dirs.
     shutDownMiniCluster();
     Configuration conf = new HdfsConfiguration();
     String[] nameDirs = new String[4];
     for (int i = 0; i < nameDirs.length; i++) {
       File nameDir = new File(System.getProperty("test.build.data"),
           "name-dir" + i);
       nameDir.mkdirs();
       nameDirs[i] = nameDir.getAbsolutePath();
     }
     conf.set(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY,
         StringUtils.join(nameDirs, ","));

     // Keep running unless there are less than 2 edits dirs remaining.
     conf.setInt(DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_KEY, 2);
     setUpMiniCluster(conf, false);

     // All journals active.
     assertTrue(doAnEdit());
     // The NN has not terminated (no ExitException thrown)

     // Invalidate 1/4 of the redundant journals.
     invalidateEditsDirAtIndex(0, false, false);
     assertTrue(doAnEdit());
     // The NN has not terminated (no ExitException thrown)

     // Invalidate 2/4 of the redundant journals.
     invalidateEditsDirAtIndex(1, false, false);
     assertTrue(doAnEdit());
     // The NN has not terminated (no ExitException thrown)

     // Invalidate 3/4 of the redundant journals.
     invalidateEditsDirAtIndex(2, false, false);

     try {
       doAnEdit();
       fail("A failure of more than the minimum number of redundant journals "
           + "should have halted ");
     } catch (RemoteException re) {
       assertTrue(re.getClassName().contains("ExitException"));
       GenericTestUtils.assertExceptionContains(
           "Could not sync enough journals to persistent storage due to " +
           "setReadyToFlush failed for too many journals. " +
           "Unsynced transactions: 1", re);
     }
   }

   /**
    * Replace the journal at index <code>index</code> with one that throws an
    * exception on flush.
    *
    * @param index the index of the journal to take offline.
    * @return the original <code>EditLogOutputStream</code> of the journal.
    */
   private void invalidateEditsDirAtIndex(int index,
       boolean failOnFlush, boolean failOnWrite) throws IOException {
     JournalAndStream jas = getJournalAndStream(index);
     EditLogFileOutputStream spyElos = spyOnStream(jas);
     if (failOnWrite) {
       doThrow(new IOException("fail on write()")).when(spyElos).write(
           (FSEditLogOp) any());
     }
     if (failOnFlush) {
       doThrow(new IOException("fail on flush()")).when(spyElos).flush();
     } else {
       doThrow(new IOException("fail on setReadyToFlush()")).when(spyElos)
         .setReadyToFlush();
     }
     doNothing().when(spyElos).abort();
   }

   private EditLogFileOutputStream spyOnStream(JournalAndStream jas) {
     EditLogFileOutputStream elos =
       (EditLogFileOutputStream) jas.getCurrentStream();
     EditLogFileOutputStream spyElos = spy(elos);
     jas.setCurrentStreamForTests(spyElos);
     return spyElos;
   }

   /**
    * Pull out one of the JournalAndStream objects from the edit log.
    */
   private JournalAndStream getJournalAndStream(int index) {
     FSImage fsimage = cluster.getNamesystem().getFSImage();
     FSEditLog editLog = fsimage.getEditLog();

     return editLog.getJournals().get(index);
   }

   /**
    * Do a mutative metadata operation on the file system.
    *
    * @return true if the operation was successful, false otherwise.
    */
   private boolean doAnEdit() throws IOException {
     return fs.mkdirs(new Path("/tmp", Integer.toString(editsPerformed++)));
   }
 }
	/**
	* 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.hadoop.hdfs.server.namenode;

	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertTrue;
	import static org.junit.Assert.fail;
	import static org.mockito.Matchers.any;
	import static org.mockito.Mockito.doNothing;
	import static org.mockito.Mockito.doThrow;
	import static org.mockito.Mockito.spy;

	import java.io.File;
	import java.io.IOException;

	import org.apache.commons.lang.StringUtils;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.hdfs.DFSConfigKeys;
	import org.apache.hadoop.hdfs.HdfsConfiguration;
	import org.apache.hadoop.hdfs.MiniDFSCluster;
	import org.apache.hadoop.hdfs.server.namenode.JournalSet.JournalAndStream;
	import org.apache.hadoop.ipc.RemoteException;
	import org.apache.hadoop.test.GenericTestUtils;
	import org.apache.hadoop.util.ExitUtil.ExitException;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.Test;
	import org.mockito.Mockito;

	public class TestEditLogJournalFailures {

	private int editsPerformed = 0;
	private MiniDFSCluster cluster;
	private FileSystem fs;

	/**
	* Create the mini cluster for testing and sub in a custom runtime so that
	* edit log journal failures don't actually cause the JVM to exit.
	*/
	@Before
	public void setUpMiniCluster() throws IOException {
	setUpMiniCluster(new HdfsConfiguration(), true);
	}

	public void setUpMiniCluster(Configuration conf, boolean manageNameDfsDirs)
	throws IOException {
	cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0)
	.manageNameDfsDirs(manageNameDfsDirs).checkExitOnShutdown(false).build();
	cluster.waitActive();
	fs = cluster.getFileSystem();
	}

	@After
	public void shutDownMiniCluster() throws IOException {
	if (fs != null)
	fs.close();
	if (cluster != null) {
	try {
	cluster.shutdown();
	} catch (ExitException ee) {
	// Ignore ExitExceptions as the tests may result in the
	// NameNode doing an immediate shutdown.
	}
	}
	}

	@Test
	public void testSingleFailedEditsDirOnFlush() throws IOException {
	assertTrue(doAnEdit());
	// Invalidate one edits journal.
	invalidateEditsDirAtIndex(0, true, false);
	// The NN has not terminated (no ExitException thrown)
	assertTrue(doAnEdit());
	// A single journal failure should not result in a call to terminate
	assertFalse(cluster.getNameNode().isInSafeMode());
	}

	@Test
	public void testAllEditsDirsFailOnFlush() throws IOException {
	assertTrue(doAnEdit());
	// Invalidate both edits journals.
	invalidateEditsDirAtIndex(0, true, false);
	invalidateEditsDirAtIndex(1, true, false);
	// The NN has not terminated (no ExitException thrown)
	try {
	doAnEdit();
	fail("The previous edit could not be synced to any persistent storage, "
	+ "should have halted the NN");
	} catch (RemoteException re) {
	assertTrue(re.getClassName().contains("ExitException"));
	GenericTestUtils.assertExceptionContains(
	"Could not sync enough journals to persistent storage. " +
	"Unsynced transactions: 1", re);
	}
	}

	@Test
	public void testAllEditsDirFailOnWrite() throws IOException {
	assertTrue(doAnEdit());
	// Invalidate both edits journals.
	invalidateEditsDirAtIndex(0, true, true);
	invalidateEditsDirAtIndex(1, true, true);
	// The NN has not terminated (no ExitException thrown)
	try {
	doAnEdit();
	fail("The previous edit could not be synced to any persistent storage, "
	+ " should have halted the NN");
	} catch (RemoteException re) {
	assertTrue(re.getClassName().contains("ExitException"));
	GenericTestUtils.assertExceptionContains(
	"Could not sync enough journals to persistent storage due to " +
	"No journals available to flush. " +
	"Unsynced transactions: 1", re);
	}
	}

	@Test
	public void testSingleFailedEditsDirOnSetReadyToFlush() throws IOException {
	assertTrue(doAnEdit());
	// Invalidate one edits journal.
	invalidateEditsDirAtIndex(0, false, false);
	// The NN has not terminated (no ExitException thrown)
	assertTrue(doAnEdit());
	// A single journal failure should not result in a call to terminate
	assertFalse(cluster.getNameNode().isInSafeMode());
	}

	@Test
	public void testSingleRequiredFailedEditsDirOnSetReadyToFlush()
	throws IOException {
	// Set one of the edits dirs to be required.
	String[] editsDirs = cluster.getConfiguration(0).getTrimmedStrings(
	DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY);
	shutDownMiniCluster();
	Configuration conf = new HdfsConfiguration();
	conf.set(DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_REQUIRED_KEY, editsDirs[0]);
	conf.setInt(DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_KEY, 0);
	conf.setInt(DFSConfigKeys.DFS_NAMENODE_CHECKED_VOLUMES_MINIMUM_KEY, 0);
	setUpMiniCluster(conf, true);

	assertTrue(doAnEdit());
	// Invalidated the one required edits journal.
	invalidateEditsDirAtIndex(0, false, false);
	JournalAndStream nonRequiredJas = getJournalAndStream(1);
	EditLogFileOutputStream nonRequiredSpy =
	spyOnStream(nonRequiredJas);

	// The NN has not terminated (no ExitException thrown)

	// ..and that the other stream is active.
	assertTrue(nonRequiredJas.isActive());

	try {
	doAnEdit();
	fail("A single failure of a required journal should have halted the NN");
	} catch (RemoteException re) {
	assertTrue(re.getClassName().contains("ExitException"));
	GenericTestUtils.assertExceptionContains(
	"setReadyToFlush failed for required journal", re);
	}

	// Since the required directory failed setReadyToFlush, and that
	// directory was listed prior to the non-required directory,
	// we should not call setReadyToFlush on the non-required
	// directory. Regression test for HDFS-2874.
	Mockito.verify(nonRequiredSpy, Mockito.never()).setReadyToFlush();
	assertFalse(nonRequiredJas.isActive());
	}

	@Test
	public void testMultipleRedundantFailedEditsDirOnSetReadyToFlush()
	throws IOException {
	// Set up 4 name/edits dirs.
	shutDownMiniCluster();
	Configuration conf = new HdfsConfiguration();
	String[] nameDirs = new String[4];
	for (int i = 0; i < nameDirs.length; i++) {
	File nameDir = new File(System.getProperty("test.build.data"),
	"name-dir" + i);
	nameDir.mkdirs();
	nameDirs[i] = nameDir.getAbsolutePath();
	}
	conf.set(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY,
	StringUtils.join(nameDirs, ","));

	// Keep running unless there are less than 2 edits dirs remaining.
	conf.setInt(DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_KEY, 2);
	setUpMiniCluster(conf, false);

	// All journals active.
	assertTrue(doAnEdit());
	// The NN has not terminated (no ExitException thrown)

	// Invalidate 1/4 of the redundant journals.
	invalidateEditsDirAtIndex(0, false, false);
	assertTrue(doAnEdit());
	// The NN has not terminated (no ExitException thrown)

	// Invalidate 2/4 of the redundant journals.
	invalidateEditsDirAtIndex(1, false, false);
	assertTrue(doAnEdit());
	// The NN has not terminated (no ExitException thrown)

	// Invalidate 3/4 of the redundant journals.
	invalidateEditsDirAtIndex(2, false, false);

	try {
	doAnEdit();
	fail("A failure of more than the minimum number of redundant journals "
	+ "should have halted ");
	} catch (RemoteException re) {
	assertTrue(re.getClassName().contains("ExitException"));
	GenericTestUtils.assertExceptionContains(
	"Could not sync enough journals to persistent storage due to " +
	"setReadyToFlush failed for too many journals. " +
	"Unsynced transactions: 1", re);
	}
	}

	/**
	* Replace the journal at index <code>index</code> with one that throws an
	* exception on flush.
	*
	* @param index the index of the journal to take offline.
	* @return the original <code>EditLogOutputStream</code> of the journal.
	*/
	private void invalidateEditsDirAtIndex(int index,
	boolean failOnFlush, boolean failOnWrite) throws IOException {
	JournalAndStream jas = getJournalAndStream(index);
	EditLogFileOutputStream spyElos = spyOnStream(jas);
	if (failOnWrite) {
	doThrow(new IOException("fail on write()")).when(spyElos).write(
	(FSEditLogOp) any());
	}
	if (failOnFlush) {
	doThrow(new IOException("fail on flush()")).when(spyElos).flush();
	} else {
	doThrow(new IOException("fail on setReadyToFlush()")).when(spyElos)
	.setReadyToFlush();
	}
	doNothing().when(spyElos).abort();
	}

	private EditLogFileOutputStream spyOnStream(JournalAndStream jas) {
	EditLogFileOutputStream elos =
	(EditLogFileOutputStream) jas.getCurrentStream();
	EditLogFileOutputStream spyElos = spy(elos);
	jas.setCurrentStreamForTests(spyElos);
	return spyElos;
	}

	/**
	* Pull out one of the JournalAndStream objects from the edit log.
	*/
	private JournalAndStream getJournalAndStream(int index) {
	FSImage fsimage = cluster.getNamesystem().getFSImage();
	FSEditLog editLog = fsimage.getEditLog();

	return editLog.getJournals().get(index);
	}

	/**
	* Do a mutative metadata operation on the file system.
	*
	* @return true if the operation was successful, false otherwise.
	*/
	private boolean doAnEdit() throws IOException {
	return fs.mkdirs(new Path("/tmp", Integer.toString(editsPerformed++)));
	}
	}