hadoop-mapreduce/src/test/mapred/org/apache/hadoop/mapred/TestShuffleExceptionCount.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.mapred;

 import static org.junit.Assert.assertEquals;

 import java.io.IOException;
 import java.util.Collection;
 import java.util.Map;

 import org.apache.hadoop.mapred.TaskTracker.ShuffleServerMetrics;
 import org.apache.hadoop.metrics.ContextFactory;
 import org.apache.hadoop.metrics.MetricsContext;
 import org.apache.hadoop.metrics.spi.OutputRecord;
 import org.junit.Test;

 public class TestShuffleExceptionCount {

   public static class TestMapOutputServlet extends TaskTracker.MapOutputServlet {

     public void checkException(IOException ie, String exceptionMsgRegex,
         String exceptionStackRegex, ShuffleServerMetrics shuffleMetrics) {
       super.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
           shuffleMetrics);
     }

   }

   @Test
   public void testCheckException() throws IOException, InterruptedException,
       ClassNotFoundException, InstantiationException, IllegalAccessException {
     TestMapOutputServlet testServlet = new TestMapOutputServlet();
     JobConf conf = new JobConf();
     conf.setUser("testuser");
     conf.setJobName("testJob");
     conf.setSessionId("testSession");

     // setup metrics context factory
     ContextFactory factory = ContextFactory.getFactory();
     factory.setAttribute("mapred.class",
         "org.apache.hadoop.metrics.spi.NoEmitMetricsContext");

     TaskTracker tt = new TaskTracker();
     tt.setConf(conf);
     ShuffleServerMetrics shuffleMetrics = tt.new ShuffleServerMetrics(conf);

     // first test with only MsgRegex set but doesn't match
     String exceptionMsgRegex = "Broken pipe";
     String exceptionStackRegex = null;
     IOException ie = new IOException("EOFException");
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);

     MetricsContext context = factory.getContext("mapred");
     shuffleMetrics.doUpdates(context);
     Map<String, Collection<OutputRecord>> records = context.getAllRecords();
     Collection<OutputRecord> col = records.get("shuffleOutput");
     OutputRecord outputRecord = col.iterator().next();
     assertEquals(0, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());

     // test with only MsgRegex set that does match
     ie = new IOException("Broken pipe");
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);

     shuffleMetrics.doUpdates(context);
     assertEquals(1, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());

     // test with neither set, make sure incremented
     exceptionStackRegex = null;
     exceptionMsgRegex = null;
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);
     shuffleMetrics.doUpdates(context);
     assertEquals(2, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());

     // test with only StackRegex set doesn't match
     exceptionStackRegex = ".*\\.doesnt\\$SelectSet\\.wakeup.*";
     exceptionMsgRegex = null;
     ie.setStackTrace(constructStackTrace());
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);
     shuffleMetrics.doUpdates(context);
     assertEquals(2, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());

     // test with only StackRegex set does match
     exceptionStackRegex = ".*\\.SelectorManager\\$SelectSet\\.wakeup.*";
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);
     shuffleMetrics.doUpdates(context);
     assertEquals(3, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());

     // test with both regex set and matches
     exceptionMsgRegex = "Broken pipe";
     ie.setStackTrace(constructStackTraceTwo());
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);
     shuffleMetrics.doUpdates(context);
     assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());

     // test with both regex set and only msg matches
     exceptionStackRegex = ".*[1-9]+BOGUSREGEX";
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);
     shuffleMetrics.doUpdates(context);
     assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());

     // test with both regex set and only stack matches
     exceptionStackRegex = ".*\\.SelectorManager\\$SelectSet\\.wakeup.*";
     exceptionMsgRegex = "EOFException";
     testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
         shuffleMetrics);
     shuffleMetrics.doUpdates(context);
     assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
         .intValue());
   }

   /*
    * Construction exception like: java.io.IOException: Broken pipe at
    * sun.nio.ch.EPollArrayWrapper.interrupt(Native Method) at
    * sun.nio.ch.EPollArrayWrapper.interrupt(EPollArrayWrapper.java:256) at
    * sun.nio.ch.EPollSelectorImpl.wakeup(EPollSelectorImpl.java:175) at
    * org.mortbay
    * .io.nio.SelectorManager$SelectSet.wakeup(SelectorManager.java:831) at
    * org.mortbay
    * .io.nio.SelectorManager$SelectSet.doSelect(SelectorManager.java:709) at
    * org.mortbay.io.nio.SelectorManager.doSelect(SelectorManager.java:192) at
    * org
    * .mortbay.jetty.nio.SelectChannelConnector.accept(SelectChannelConnector.java
    * :124) at
    * org.mortbay.jetty.AbstractConnector$Acceptor.run(AbstractConnector.
    * java:708) at
    * org.mortbay.thread.QueuedThreadPool$PoolThread.run(QueuedThreadPool
    * .java:582)
    */
   private StackTraceElement[] constructStackTrace() {
     StackTraceElement[] stack = new StackTraceElement[9];
     stack[0] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
         "interrupt", "", -2);
     stack[1] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
         "interrupt", "EPollArrayWrapper.java", 256);
     stack[2] = new StackTraceElement("sun.nio.ch.EPollSelectorImpl", "wakeup",
         "EPollSelectorImpl.java", 175);
     stack[3] = new StackTraceElement(
         "org.mortbay.io.nio.SelectorManager$SelectSet", "wakeup",
         "SelectorManager.java", 831);
     stack[4] = new StackTraceElement(
         "org.mortbay.io.nio.SelectorManager$SelectSet", "doSelect",
         "SelectorManager.java", 709);
     stack[5] = new StackTraceElement("org.mortbay.io.nio.SelectorManager",
         "doSelect", "SelectorManager.java", 192);
     stack[6] = new StackTraceElement(
         "org.mortbay.jetty.nio.SelectChannelConnector", "accept",
         "SelectChannelConnector.java", 124);
     stack[7] = new StackTraceElement(
         "org.mortbay.jetty.AbstractConnector$Acceptor", "run",
         "AbstractConnector.java", 708);
     stack[8] = new StackTraceElement(
         "org.mortbay.thread.QueuedThreadPool$PoolThread", "run",
         "QueuedThreadPool.java", 582);

     return stack;
   }

   /*
    * java.io.IOException: Broken pipe at
    * sun.nio.ch.EPollArrayWrapper.interrupt(Native Method) at
    * sun.nio.ch.EPollArrayWrapper.interrupt(EPollArrayWrapper.java:256) at
    * sun.nio.ch.EPollSelectorImpl.wakeup(EPollSelectorImpl.java:175) at
    * org.mortbay
    * .io.nio.SelectorManager$SelectSet.wakeup(SelectorManager.java:831) at
    * org.mortbay
    * .io.nio.SelectChannelEndPoint.updateKey(SelectChannelEndPoint.java:335) at
    * org
    * .mortbay.io.nio.SelectChannelEndPoint.blockWritable(SelectChannelEndPoint
    * .java:278) at
    * org.mortbay.jetty.AbstractGenerator$Output.blockForOutput(AbstractGenerator
    * .java:545) at
    * org.mortbay.jetty.AbstractGenerator$Output.flush(AbstractGenerator
    * .java:572) at
    * org.mortbay.jetty.HttpConnection$Output.flush(HttpConnection.java:1012) at
    * org
    * .mortbay.jetty.AbstractGenerator$Output.write(AbstractGenerator.java:651)at
    * org
    * .mortbay.jetty.AbstractGenerator$Output.write(AbstractGenerator.java:580)
    * at
    */
   private StackTraceElement[] constructStackTraceTwo() {
     StackTraceElement[] stack = new StackTraceElement[11];
     stack[0] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
         "interrupt", "", -2);
     stack[1] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
         "interrupt", "EPollArrayWrapper.java", 256);
     stack[2] = new StackTraceElement("sun.nio.ch.EPollSelectorImpl", "wakeup",
         "EPollSelectorImpl.java", 175);
     stack[3] = new StackTraceElement(
         "org.mortbay.io.nio.SelectorManager$SelectSet", "wakeup",
         "SelectorManager.java", 831);
     stack[4] = new StackTraceElement(
         "org.mortbay.io.nio.SelectChannelEndPoint", "updateKey",
         "SelectChannelEndPoint.java", 335);
     stack[5] = new StackTraceElement(
         "org.mortbay.io.nio.SelectChannelEndPoint", "blockWritable",
         "SelectChannelEndPoint.java", 278);
     stack[6] = new StackTraceElement(
         "org.mortbay.jetty.AbstractGenerator$Output", "blockForOutput",
         "AbstractGenerator.java", 545);
     stack[7] = new StackTraceElement(
         "org.mortbay.jetty.AbstractGenerator$Output", "flush",
         "AbstractGenerator.java", 572);
     stack[8] = new StackTraceElement("org.mortbay.jetty.HttpConnection$Output",
         "flush", "HttpConnection.java", 1012);
     stack[9] = new StackTraceElement(
         "org.mortbay.jetty.AbstractGenerator$Output", "write",
         "AbstractGenerator.java", 651);
     stack[10] = new StackTraceElement(
         "org.mortbay.jetty.AbstractGenerator$Output", "write",
         "AbstractGenerator.java", 580);

     return stack;
   }

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

	import static org.junit.Assert.assertEquals;

	import java.io.IOException;
	import java.util.Collection;
	import java.util.Map;

	import org.apache.hadoop.mapred.TaskTracker.ShuffleServerMetrics;
	import org.apache.hadoop.metrics.ContextFactory;
	import org.apache.hadoop.metrics.MetricsContext;
	import org.apache.hadoop.metrics.spi.OutputRecord;
	import org.junit.Test;

	public class TestShuffleExceptionCount {

	public static class TestMapOutputServlet extends TaskTracker.MapOutputServlet {

	public void checkException(IOException ie, String exceptionMsgRegex,
	String exceptionStackRegex, ShuffleServerMetrics shuffleMetrics) {
	super.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);
	}

	}

	@Test
	public void testCheckException() throws IOException, InterruptedException,
	ClassNotFoundException, InstantiationException, IllegalAccessException {
	TestMapOutputServlet testServlet = new TestMapOutputServlet();
	JobConf conf = new JobConf();
	conf.setUser("testuser");
	conf.setJobName("testJob");
	conf.setSessionId("testSession");

	// setup metrics context factory
	ContextFactory factory = ContextFactory.getFactory();
	factory.setAttribute("mapred.class",
	"org.apache.hadoop.metrics.spi.NoEmitMetricsContext");

	TaskTracker tt = new TaskTracker();
	tt.setConf(conf);
	ShuffleServerMetrics shuffleMetrics = tt.new ShuffleServerMetrics(conf);

	// first test with only MsgRegex set but doesn't match
	String exceptionMsgRegex = "Broken pipe";
	String exceptionStackRegex = null;
	IOException ie = new IOException("EOFException");
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);

	MetricsContext context = factory.getContext("mapred");
	shuffleMetrics.doUpdates(context);
	Map<String, Collection<OutputRecord>> records = context.getAllRecords();
	Collection<OutputRecord> col = records.get("shuffleOutput");
	OutputRecord outputRecord = col.iterator().next();
	assertEquals(0, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());

	// test with only MsgRegex set that does match
	ie = new IOException("Broken pipe");
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);

	shuffleMetrics.doUpdates(context);
	assertEquals(1, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());

	// test with neither set, make sure incremented
	exceptionStackRegex = null;
	exceptionMsgRegex = null;
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);
	shuffleMetrics.doUpdates(context);
	assertEquals(2, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());

	// test with only StackRegex set doesn't match
	exceptionStackRegex = ".\\.doesnt\\$SelectSet\\.wakeup.";
	exceptionMsgRegex = null;
	ie.setStackTrace(constructStackTrace());
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);
	shuffleMetrics.doUpdates(context);
	assertEquals(2, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());

	// test with only StackRegex set does match
	exceptionStackRegex = ".\\.SelectorManager\\$SelectSet\\.wakeup.";
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);
	shuffleMetrics.doUpdates(context);
	assertEquals(3, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());

	// test with both regex set and matches
	exceptionMsgRegex = "Broken pipe";
	ie.setStackTrace(constructStackTraceTwo());
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);
	shuffleMetrics.doUpdates(context);
	assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());

	// test with both regex set and only msg matches
	exceptionStackRegex = ".*[1-9]+BOGUSREGEX";
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);
	shuffleMetrics.doUpdates(context);
	assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());

	// test with both regex set and only stack matches
	exceptionStackRegex = ".\\.SelectorManager\\$SelectSet\\.wakeup.";
	exceptionMsgRegex = "EOFException";
	testServlet.checkException(ie, exceptionMsgRegex, exceptionStackRegex,
	shuffleMetrics);
	shuffleMetrics.doUpdates(context);
	assertEquals(4, outputRecord.getMetric("shuffle_exceptions_caught")
	.intValue());
	}

	/*
	* Construction exception like: java.io.IOException: Broken pipe at
	* sun.nio.ch.EPollArrayWrapper.interrupt(Native Method) at
	* sun.nio.ch.EPollArrayWrapper.interrupt(EPollArrayWrapper.java:256) at
	* sun.nio.ch.EPollSelectorImpl.wakeup(EPollSelectorImpl.java:175) at
	* org.mortbay
	* .io.nio.SelectorManager$SelectSet.wakeup(SelectorManager.java:831) at
	* org.mortbay
	* .io.nio.SelectorManager$SelectSet.doSelect(SelectorManager.java:709) at
	* org.mortbay.io.nio.SelectorManager.doSelect(SelectorManager.java:192) at
	* org
	* .mortbay.jetty.nio.SelectChannelConnector.accept(SelectChannelConnector.java
	* :124) at
	* org.mortbay.jetty.AbstractConnector$Acceptor.run(AbstractConnector.
	* java:708) at
	* org.mortbay.thread.QueuedThreadPool$PoolThread.run(QueuedThreadPool
	* .java:582)
	*/
	private StackTraceElement[] constructStackTrace() {
	StackTraceElement[] stack = new StackTraceElement[9];
	stack[0] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
	"interrupt", "", -2);
	stack[1] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
	"interrupt", "EPollArrayWrapper.java", 256);
	stack[2] = new StackTraceElement("sun.nio.ch.EPollSelectorImpl", "wakeup",
	"EPollSelectorImpl.java", 175);
	stack[3] = new StackTraceElement(
	"org.mortbay.io.nio.SelectorManager$SelectSet", "wakeup",
	"SelectorManager.java", 831);
	stack[4] = new StackTraceElement(
	"org.mortbay.io.nio.SelectorManager$SelectSet", "doSelect",
	"SelectorManager.java", 709);
	stack[5] = new StackTraceElement("org.mortbay.io.nio.SelectorManager",
	"doSelect", "SelectorManager.java", 192);
	stack[6] = new StackTraceElement(
	"org.mortbay.jetty.nio.SelectChannelConnector", "accept",
	"SelectChannelConnector.java", 124);
	stack[7] = new StackTraceElement(
	"org.mortbay.jetty.AbstractConnector$Acceptor", "run",
	"AbstractConnector.java", 708);
	stack[8] = new StackTraceElement(
	"org.mortbay.thread.QueuedThreadPool$PoolThread", "run",
	"QueuedThreadPool.java", 582);

	return stack;
	}

	/*
	* java.io.IOException: Broken pipe at
	* sun.nio.ch.EPollArrayWrapper.interrupt(Native Method) at
	* sun.nio.ch.EPollArrayWrapper.interrupt(EPollArrayWrapper.java:256) at
	* sun.nio.ch.EPollSelectorImpl.wakeup(EPollSelectorImpl.java:175) at
	* org.mortbay
	* .io.nio.SelectorManager$SelectSet.wakeup(SelectorManager.java:831) at
	* org.mortbay
	* .io.nio.SelectChannelEndPoint.updateKey(SelectChannelEndPoint.java:335) at
	* org
	* .mortbay.io.nio.SelectChannelEndPoint.blockWritable(SelectChannelEndPoint
	* .java:278) at
	* org.mortbay.jetty.AbstractGenerator$Output.blockForOutput(AbstractGenerator
	* .java:545) at
	* org.mortbay.jetty.AbstractGenerator$Output.flush(AbstractGenerator
	* .java:572) at
	* org.mortbay.jetty.HttpConnection$Output.flush(HttpConnection.java:1012) at
	* org
	* .mortbay.jetty.AbstractGenerator$Output.write(AbstractGenerator.java:651)at
	* org
	* .mortbay.jetty.AbstractGenerator$Output.write(AbstractGenerator.java:580)
	* at
	*/
	private StackTraceElement[] constructStackTraceTwo() {
	StackTraceElement[] stack = new StackTraceElement[11];
	stack[0] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
	"interrupt", "", -2);
	stack[1] = new StackTraceElement("sun.nio.ch.EPollArrayWrapper",
	"interrupt", "EPollArrayWrapper.java", 256);
	stack[2] = new StackTraceElement("sun.nio.ch.EPollSelectorImpl", "wakeup",
	"EPollSelectorImpl.java", 175);
	stack[3] = new StackTraceElement(
	"org.mortbay.io.nio.SelectorManager$SelectSet", "wakeup",
	"SelectorManager.java", 831);
	stack[4] = new StackTraceElement(
	"org.mortbay.io.nio.SelectChannelEndPoint", "updateKey",
	"SelectChannelEndPoint.java", 335);
	stack[5] = new StackTraceElement(
	"org.mortbay.io.nio.SelectChannelEndPoint", "blockWritable",
	"SelectChannelEndPoint.java", 278);
	stack[6] = new StackTraceElement(
	"org.mortbay.jetty.AbstractGenerator$Output", "blockForOutput",
	"AbstractGenerator.java", 545);
	stack[7] = new StackTraceElement(
	"org.mortbay.jetty.AbstractGenerator$Output", "flush",
	"AbstractGenerator.java", 572);
	stack[8] = new StackTraceElement("org.mortbay.jetty.HttpConnection$Output",
	"flush", "HttpConnection.java", 1012);
	stack[9] = new StackTraceElement(
	"org.mortbay.jetty.AbstractGenerator$Output", "write",
	"AbstractGenerator.java", 651);
	stack[10] = new StackTraceElement(
	"org.mortbay.jetty.AbstractGenerator$Output", "write",
	"AbstractGenerator.java", 580);

	return stack;
	}

	}