hdfs/src/test/aop/org/apache/hadoop/hdfs/TestFiPipelines.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;

 import static org.junit.Assert.assertTrue;

 import java.io.IOException;
 import java.util.Random;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.logging.impl.Log4JLogger;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fi.FiTestUtil;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.server.datanode.BlockReceiverAspects;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.log4j.Level;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;

 public class TestFiPipelines {
   public static final Log LOG = LogFactory.getLog(TestFiPipelines.class);

   private static short REPL_FACTOR = 3;
   private static final int RAND_LIMIT = 2000;

   private MiniDFSCluster cluster;
   private DistributedFileSystem fs;
   private static Configuration conf;
   Random rand = new Random(RAND_LIMIT);

   static {
     initLoggers();
     setConfiguration();
   }

   @Before
   public void startUpCluster() throws IOException {
     cluster = new MiniDFSCluster(conf, REPL_FACTOR, true, null);
     fs = (DistributedFileSystem) cluster.getFileSystem();
   }

   @After
   synchronized public void shutDownCluster() throws IOException {
     if (cluster != null) cluster.shutdown();
   }

   /**
    * Test initiates and sets actions created by injection framework. The actions
    * work with both aspects of sending acknologment packets in a pipeline.
    * Creates and closes a file of certain length < packet size.
    * Injected actions will check if number of visible bytes at datanodes equals
    * to number of acknoleged bytes
    *
    * @throws IOException in case of an error
    */
   @Test
   public void pipeline_04() throws IOException {
     final String METHOD_NAME = GenericTestUtils.getMethodName();
     LOG.debug("Running " + METHOD_NAME);

     final PipelinesTestUtil.PipelinesTest pipst =
       (PipelinesTestUtil.PipelinesTest) PipelinesTestUtil.initTest();

     pipst.fiCallSetNumBytes.set(new PipelinesTestUtil.ReceivedCheckAction(METHOD_NAME));
     pipst.fiCallSetBytesAcked.set(new PipelinesTestUtil.AckedCheckAction(METHOD_NAME));

     Path filePath = new Path("/" + METHOD_NAME + ".dat");
     FSDataOutputStream fsOut = fs.create(filePath);
     TestPipelines.writeData(fsOut, 2);
     fs.close();
   }

   /**
    * Similar to pipeline_04 but sends many packets into a pipeline
    * @throws IOException in case of an error
    */
   @Test
   public void pipeline_05() throws IOException {
     final String METHOD_NAME = GenericTestUtils.getMethodName();
     LOG.debug("Running " + METHOD_NAME);

     final PipelinesTestUtil.PipelinesTest pipst =
       (PipelinesTestUtil.PipelinesTest) PipelinesTestUtil.initTest();

     pipst.fiCallSetNumBytes.set(new PipelinesTestUtil.ReceivedCheckAction(METHOD_NAME));
     pipst.fiCallSetBytesAcked.set(new PipelinesTestUtil.AckedCheckAction(METHOD_NAME));

     Path filePath = new Path("/" + METHOD_NAME + ".dat");
     FSDataOutputStream fsOut = fs.create(filePath);
     for (int i = 0; i < 17; i++) {
       TestPipelines.writeData(fsOut, 23);
     }
     fs.close();
   }

   /**
    * This quite tricky test prevents acknowledgement packets from a datanode
    * This should block any write attempts after ackQueue is full.
    * Test is blocking, so the MiniDFSCluster has to be killed harshly.
    * @throws IOException in case of an error
    */
   @Test
   public void pipeline_06() throws IOException {
     final String METHOD_NAME = GenericTestUtils.getMethodName();
     final int MAX_PACKETS = 80;

     LOG.debug("Running " + METHOD_NAME);

     final PipelinesTestUtil.PipelinesTest pipst =
       (PipelinesTestUtil.PipelinesTest) PipelinesTestUtil.initTest();

     pipst.setSuspend(true); // This is ack. suspend test
     Path filePath = new Path("/" + METHOD_NAME + ".dat");
     FSDataOutputStream fsOut = fs.create(filePath);

     int cnt = 0;
     try {
       // At this point let's start an external checker thread, which will
       // verify the test's results and shutdown the MiniDFSCluster for us,
       // because what it's gonna do has BLOCKING effect on datanodes
       QueueChecker cq = new QueueChecker(pipst, MAX_PACKETS);
       cq.start();
       // The following value is explained by the fact that size of a packet isn't
       // necessary equals to the value of
       // DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_KEY
       // The actual logic is expressed in DFSClient#computePacketChunkSize
       int bytesToSend = 700;
       while (cnt < 100 && pipst.getSuspend()) {
         LOG.debug("_06(): " + cnt++ + " sending another " + bytesToSend + " bytes");
         TestPipelines.writeData(fsOut, bytesToSend);
       }
     } catch (Exception e) {
       LOG.warn("Getting unexpected exception: ", e);
     }
     LOG.debug("Last queued packet number " + pipst.getLastQueued());
     assertTrue("Shouldn't be able to send more than 81 packet", pipst.getLastQueued() <= 81);
   }

   private class QueueChecker extends Thread {
     PipelinesTestUtil.PipelinesTest test;
     final int MAX;
     boolean done = false;

     public QueueChecker(PipelinesTestUtil.PipelinesTest handle, int maxPackets) {
       test = handle;
       MAX = maxPackets;
     }

     @Override
     public void run() {
       while (!done) {
         LOG.debug("_06: checking for the limit " + test.getLastQueued() +
             " and " + MAX);
         if (test.getLastQueued() >= MAX) {
           LOG.debug("FI: Resume packets acking");
           test.setSuspend(false); //Do not suspend ack sending any more
           done = true;
         }
         if (!done)
           try {
             LOG.debug("_06: MAX isn't reached yet. Current=" + test.getLastQueued());
             sleep(100);
           } catch (InterruptedException e) { }
       }

       assertTrue("Shouldn't be able to send more than 81 packet", test.getLastQueued() <= 81);
       try {
         LOG.debug("_06: shutting down the cluster");
         // It has to be done like that, because local version of shutDownCluster()
         // won't work, because it tries to close an instance of FileSystem too.
         // Which is where the waiting is happening.
         if (cluster !=null )
           shutDownCluster();
       } catch (Exception e) {
         e.printStackTrace();
       }
       LOG.debug("End QueueChecker thread");
     }
   }

   private static void setConfiguration() {
     conf = new Configuration();
     int customPerChecksumSize = 700;
     int customBlockSize = customPerChecksumSize * 3;
     conf.setInt(DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_KEY, 100);
     conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, customPerChecksumSize);
     conf.setInt(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, customBlockSize);
     conf.setInt(DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_KEY, customBlockSize / 2);
     conf.setBoolean(DFSConfigKeys.DFS_SUPPORT_APPEND_KEY, true);
     conf.setInt(DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY, 0);
   }

   private static void initLoggers() {
     ((Log4JLogger) NameNode.stateChangeLog).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) FSNamesystem.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) DataNode.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) TestFiPipelines.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) DFSClient.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) FiTestUtil.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) BlockReceiverAspects.LOG).getLogger().setLevel(Level.ALL);
     ((Log4JLogger) DFSClientAspects.LOG).getLogger().setLevel(Level.ALL);
   }

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

	import static org.junit.Assert.assertTrue;

	import java.io.IOException;
	import java.util.Random;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.commons.logging.impl.Log4JLogger;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fi.FiTestUtil;
	import org.apache.hadoop.fs.FSDataOutputStream;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.hdfs.server.datanode.BlockReceiverAspects;
	import org.apache.hadoop.hdfs.server.datanode.DataNode;
	import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
	import org.apache.hadoop.hdfs.server.namenode.NameNode;
	import org.apache.hadoop.test.GenericTestUtils;
	import org.apache.log4j.Level;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.Test;

	public class TestFiPipelines {
	public static final Log LOG = LogFactory.getLog(TestFiPipelines.class);

	private static short REPL_FACTOR = 3;
	private static final int RAND_LIMIT = 2000;

	private MiniDFSCluster cluster;
	private DistributedFileSystem fs;
	private static Configuration conf;
	Random rand = new Random(RAND_LIMIT);

	static {
	initLoggers();
	setConfiguration();
	}

	@Before
	public void startUpCluster() throws IOException {
	cluster = new MiniDFSCluster(conf, REPL_FACTOR, true, null);
	fs = (DistributedFileSystem) cluster.getFileSystem();
	}

	@After
	synchronized public void shutDownCluster() throws IOException {
	if (cluster != null) cluster.shutdown();
	}

	/**
	* Test initiates and sets actions created by injection framework. The actions
	* work with both aspects of sending acknologment packets in a pipeline.
	* Creates and closes a file of certain length < packet size.
	* Injected actions will check if number of visible bytes at datanodes equals
	* to number of acknoleged bytes
	*
	* @throws IOException in case of an error
	*/
	@Test
	public void pipeline_04() throws IOException {
	final String METHOD_NAME = GenericTestUtils.getMethodName();
	LOG.debug("Running " + METHOD_NAME);

	final PipelinesTestUtil.PipelinesTest pipst =
	(PipelinesTestUtil.PipelinesTest) PipelinesTestUtil.initTest();

	pipst.fiCallSetNumBytes.set(new PipelinesTestUtil.ReceivedCheckAction(METHOD_NAME));
	pipst.fiCallSetBytesAcked.set(new PipelinesTestUtil.AckedCheckAction(METHOD_NAME));

	Path filePath = new Path("/" + METHOD_NAME + ".dat");
	FSDataOutputStream fsOut = fs.create(filePath);
	TestPipelines.writeData(fsOut, 2);
	fs.close();
	}

	/**
	* Similar to pipeline_04 but sends many packets into a pipeline
	* @throws IOException in case of an error
	*/
	@Test
	public void pipeline_05() throws IOException {
	final String METHOD_NAME = GenericTestUtils.getMethodName();
	LOG.debug("Running " + METHOD_NAME);

	final PipelinesTestUtil.PipelinesTest pipst =
	(PipelinesTestUtil.PipelinesTest) PipelinesTestUtil.initTest();

	pipst.fiCallSetNumBytes.set(new PipelinesTestUtil.ReceivedCheckAction(METHOD_NAME));
	pipst.fiCallSetBytesAcked.set(new PipelinesTestUtil.AckedCheckAction(METHOD_NAME));

	Path filePath = new Path("/" + METHOD_NAME + ".dat");
	FSDataOutputStream fsOut = fs.create(filePath);
	for (int i = 0; i < 17; i++) {
	TestPipelines.writeData(fsOut, 23);
	}
	fs.close();
	}

	/**
	* This quite tricky test prevents acknowledgement packets from a datanode
	* This should block any write attempts after ackQueue is full.
	* Test is blocking, so the MiniDFSCluster has to be killed harshly.
	* @throws IOException in case of an error
	*/
	@Test
	public void pipeline_06() throws IOException {
	final String METHOD_NAME = GenericTestUtils.getMethodName();
	final int MAX_PACKETS = 80;

	LOG.debug("Running " + METHOD_NAME);

	final PipelinesTestUtil.PipelinesTest pipst =
	(PipelinesTestUtil.PipelinesTest) PipelinesTestUtil.initTest();

	pipst.setSuspend(true); // This is ack. suspend test
	Path filePath = new Path("/" + METHOD_NAME + ".dat");
	FSDataOutputStream fsOut = fs.create(filePath);

	int cnt = 0;
	try {
	// At this point let's start an external checker thread, which will
	// verify the test's results and shutdown the MiniDFSCluster for us,
	// because what it's gonna do has BLOCKING effect on datanodes
	QueueChecker cq = new QueueChecker(pipst, MAX_PACKETS);
	cq.start();
	// The following value is explained by the fact that size of a packet isn't
	// necessary equals to the value of
	// DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_KEY
	// The actual logic is expressed in DFSClient#computePacketChunkSize
	int bytesToSend = 700;
	while (cnt < 100 && pipst.getSuspend()) {
	LOG.debug("_06(): " + cnt++ + " sending another " + bytesToSend + " bytes");
	TestPipelines.writeData(fsOut, bytesToSend);
	}
	} catch (Exception e) {
	LOG.warn("Getting unexpected exception: ", e);
	}
	LOG.debug("Last queued packet number " + pipst.getLastQueued());
	assertTrue("Shouldn't be able to send more than 81 packet", pipst.getLastQueued() <= 81);
	}

	private class QueueChecker extends Thread {
	PipelinesTestUtil.PipelinesTest test;
	final int MAX;
	boolean done = false;

	public QueueChecker(PipelinesTestUtil.PipelinesTest handle, int maxPackets) {
	test = handle;
	MAX = maxPackets;
	}

	@Override
	public void run() {
	while (!done) {
	LOG.debug("_06: checking for the limit " + test.getLastQueued() +
	" and " + MAX);
	if (test.getLastQueued() >= MAX) {
	LOG.debug("FI: Resume packets acking");
	test.setSuspend(false); //Do not suspend ack sending any more
	done = true;
	}
	if (!done)
	try {
	LOG.debug("_06: MAX isn't reached yet. Current=" + test.getLastQueued());
	sleep(100);
	} catch (InterruptedException e) { }
	}

	assertTrue("Shouldn't be able to send more than 81 packet", test.getLastQueued() <= 81);
	try {
	LOG.debug("_06: shutting down the cluster");
	// It has to be done like that, because local version of shutDownCluster()
	// won't work, because it tries to close an instance of FileSystem too.
	// Which is where the waiting is happening.
	if (cluster !=null )
	shutDownCluster();
	} catch (Exception e) {
	e.printStackTrace();
	}
	LOG.debug("End QueueChecker thread");
	}
	}

	private static void setConfiguration() {
	conf = new Configuration();
	int customPerChecksumSize = 700;
	int customBlockSize = customPerChecksumSize * 3;
	conf.setInt(DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_KEY, 100);
	conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, customPerChecksumSize);
	conf.setInt(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, customBlockSize);
	conf.setInt(DFSConfigKeys.DFS_CLIENT_WRITE_PACKET_SIZE_KEY, customBlockSize / 2);
	conf.setBoolean(DFSConfigKeys.DFS_SUPPORT_APPEND_KEY, true);
	conf.setInt(DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY, 0);
	}

	private static void initLoggers() {
	((Log4JLogger) NameNode.stateChangeLog).getLogger().setLevel(Level.ALL);
	((Log4JLogger) FSNamesystem.LOG).getLogger().setLevel(Level.ALL);
	((Log4JLogger) DataNode.LOG).getLogger().setLevel(Level.ALL);
	((Log4JLogger) TestFiPipelines.LOG).getLogger().setLevel(Level.ALL);
	((Log4JLogger) DFSClient.LOG).getLogger().setLevel(Level.ALL);
	((Log4JLogger) FiTestUtil.LOG).getLogger().setLevel(Level.ALL);
	((Log4JLogger) BlockReceiverAspects.LOG).getLogger().setLevel(Level.ALL);
	((Log4JLogger) DFSClientAspects.LOG).getLogger().setLevel(Level.ALL);
	}

	}