diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
index 184baaa3ff6..45ddb9eb60a 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
@@ -62,6 +62,7 @@
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent;
 import org.apache.hadoop.yarn.client.api.TimelineClient;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.event.EventHandler;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
@@ -243,9 +244,15 @@ protected void serviceInit(Configuration conf) throws Exception {
 
     if (conf.getBoolean(MRJobConfig.MAPREDUCE_JOB_EMIT_TIMELINE_DATA,
         MRJobConfig.DEFAULT_MAPREDUCE_JOB_EMIT_TIMELINE_DATA)) {
-      timelineClient = TimelineClient.createTimelineClient();
-      timelineClient.init(conf);
-      LOG.info("Emitting job history data to the timeline server is enabled");
+      if (conf.getBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED,
+            YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED)) {
+        timelineClient = TimelineClient.createTimelineClient();
+        timelineClient.init(conf);
+        LOG.info("Timeline service is enabled");
+        LOG.info("Emitting job history data to the timeline server is enabled");
+      } else {
+        LOG.info("Timeline service is not enabled");
+      }
     } else {
       LOG.info("Emitting job history data to the timeline server is not enabled");
     }
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index deb40726133..6d86d7583ae 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -81,6 +81,9 @@ Release 2.7.0 - UNRELEASED
     YARN-2802. ClusterMetrics to include AM launch and register delays.
     (Zhihai Xu via kasha)
 
+    YARN-2375. Allow enabling/disabling timeline server per framework.
+    (Mit Desai via jeagles)
+
   OPTIMIZATIONS
 
   BUG FIXES
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
index 7906e6fd001..b3ea8659fc9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
@@ -497,10 +497,16 @@ public boolean init(String[] args) throws ParseException, IOException {
     requestPriority = Integer.parseInt(cliParser
         .getOptionValue("priority", "0"));
 
-    // Creating the Timeline Client
-    timelineClient = TimelineClient.createTimelineClient();
-    timelineClient.init(conf);
-    timelineClient.start();
+    if (conf.getBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED,
+      YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED)) {
+      // Creating the Timeline Client
+      timelineClient = TimelineClient.createTimelineClient();
+      timelineClient.init(conf);
+      timelineClient.start();
+    } else {
+      timelineClient = null;
+      LOG.warn("Timeline service is not enabled");
+    }
 
     return true;
   }
@@ -548,9 +554,11 @@ public void run() throws YarnException, IOException {
     appSubmitterUgi =
         UserGroupInformation.createRemoteUser(appSubmitterUserName);
     appSubmitterUgi.addCredentials(credentials);
-
-    publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
-        DSEvent.DS_APP_ATTEMPT_START, domainId, appSubmitterUgi);
+    
+    if(timelineClient != null) {
+      publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
+          DSEvent.DS_APP_ATTEMPT_START, domainId, appSubmitterUgi);
+    }
 
     AMRMClientAsync.CallbackHandler allocListener = new RMCallbackHandler();
     amRMClient = AMRMClientAsync.createAMRMClientAsync(1000, allocListener);
@@ -617,8 +625,10 @@ public void run() throws YarnException, IOException {
     }
     numRequestedContainers.set(numTotalContainers);
 
-    publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
-        DSEvent.DS_APP_ATTEMPT_END, domainId, appSubmitterUgi);
+    if(timelineClient != null) {
+      publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
+          DSEvent.DS_APP_ATTEMPT_END, domainId, appSubmitterUgi);
+    }
   }
 
   @VisibleForTesting
@@ -681,6 +691,11 @@ protected boolean finish() {
     
     amRMClient.stop();
 
+    // Stop Timeline Client
+    if(timelineClient != null) {
+      timelineClient.stop();
+    }
+
     return success;
   }
   
@@ -724,8 +739,10 @@ public void onContainersCompleted(List<ContainerStatus> completedContainers) {
           LOG.info("Container completed successfully." + ", containerId="
               + containerStatus.getContainerId());
         }
-        publishContainerEndEvent(
-            timelineClient, containerStatus, domainId, appSubmitterUgi);
+        if(timelineClient != null) {
+          publishContainerEndEvent(
+              timelineClient, containerStatus, domainId, appSubmitterUgi);
+        }
       }
       
       // ask for more containers if any failed
@@ -840,9 +857,11 @@ public void onContainerStarted(ContainerId containerId,
       if (container != null) {
         applicationMaster.nmClientAsync.getContainerStatusAsync(containerId, container.getNodeId());
       }
-      ApplicationMaster.publishContainerStartEvent(
-          applicationMaster.timelineClient, container,
-          applicationMaster.domainId, applicationMaster.appSubmitterUgi);
+      if(applicationMaster.timelineClient != null) {
+        ApplicationMaster.publishContainerStartEvent(
+            applicationMaster.timelineClient, container,
+            applicationMaster.domainId, applicationMaster.appSubmitterUgi);
+      }
     }
 
     @Override
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java
index 2028cc95fa3..78901c365fe 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java
@@ -105,7 +105,6 @@ public class TimelineClientImpl extends TimelineClient {
   private DelegationTokenAuthenticator authenticator;
   private DelegationTokenAuthenticatedURL.Token token;
   private URI resURI;
-  private boolean isEnabled;
 
   @Private
   @VisibleForTesting
@@ -247,55 +246,42 @@ public TimelineClientImpl() {
   }
 
   protected void serviceInit(Configuration conf) throws Exception {
-    isEnabled = conf.getBoolean(
-        YarnConfiguration.TIMELINE_SERVICE_ENABLED,
-        YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED);
-    if (!isEnabled) {
-      LOG.info("Timeline service is not enabled");
+    ClientConfig cc = new DefaultClientConfig();
+    cc.getClasses().add(YarnJacksonJaxbJsonProvider.class);
+    connConfigurator = newConnConfigurator(conf);
+    if (UserGroupInformation.isSecurityEnabled()) {
+      authenticator = new KerberosDelegationTokenAuthenticator();
     } else {
-      ClientConfig cc = new DefaultClientConfig();
-      cc.getClasses().add(YarnJacksonJaxbJsonProvider.class);
-      connConfigurator = newConnConfigurator(conf);
-      if (UserGroupInformation.isSecurityEnabled()) {
-        authenticator = new KerberosDelegationTokenAuthenticator();
-      } else {
-        authenticator = new PseudoDelegationTokenAuthenticator();
-      }
-      authenticator.setConnectionConfigurator(connConfigurator);
-      token = new DelegationTokenAuthenticatedURL.Token();
-
-      connectionRetry = new TimelineClientConnectionRetry(conf);
-      client = new Client(new URLConnectionClientHandler(
-          new TimelineURLConnectionFactory()), cc);
-      TimelineJerseyRetryFilter retryFilter = new TimelineJerseyRetryFilter();
-      client.addFilter(retryFilter);
-
-      if (YarnConfiguration.useHttps(conf)) {
-        resURI = URI
-            .create(JOINER.join("https://", conf.get(
-                YarnConfiguration.TIMELINE_SERVICE_WEBAPP_HTTPS_ADDRESS,
-                YarnConfiguration.DEFAULT_TIMELINE_SERVICE_WEBAPP_HTTPS_ADDRESS),
-                RESOURCE_URI_STR));
-      } else {
-        resURI = URI.create(JOINER.join("http://", conf.get(
-            YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS,
-            YarnConfiguration.DEFAULT_TIMELINE_SERVICE_WEBAPP_ADDRESS),
-            RESOURCE_URI_STR));
-      }
-      LOG.info("Timeline service address: " + resURI);
+      authenticator = new PseudoDelegationTokenAuthenticator();
     }
+    authenticator.setConnectionConfigurator(connConfigurator);
+    token = new DelegationTokenAuthenticatedURL.Token();
+
+    connectionRetry = new TimelineClientConnectionRetry(conf);
+    client = new Client(new URLConnectionClientHandler(
+        new TimelineURLConnectionFactory()), cc);
+    TimelineJerseyRetryFilter retryFilter = new TimelineJerseyRetryFilter();
+    client.addFilter(retryFilter);
+
+    if (YarnConfiguration.useHttps(conf)) {
+      resURI = URI
+          .create(JOINER.join("https://", conf.get(
+              YarnConfiguration.TIMELINE_SERVICE_WEBAPP_HTTPS_ADDRESS,
+              YarnConfiguration.DEFAULT_TIMELINE_SERVICE_WEBAPP_HTTPS_ADDRESS),
+              RESOURCE_URI_STR));
+    } else {
+      resURI = URI.create(JOINER.join("http://", conf.get(
+          YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS,
+          YarnConfiguration.DEFAULT_TIMELINE_SERVICE_WEBAPP_ADDRESS),
+          RESOURCE_URI_STR));
+    }
+    LOG.info("Timeline service address: " + resURI);
     super.serviceInit(conf);
   }
 
   @Override
   public TimelinePutResponse putEntities(
       TimelineEntity... entities) throws IOException, YarnException {
-    if (!isEnabled) {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Nothing will be put because timeline service is not enabled");
-      }
-      return new TimelinePutResponse();
-    }
     TimelineEntities entitiesContainer = new TimelineEntities();
     entitiesContainer.addEntities(Arrays.asList(entities));
     ClientResponse resp = doPosting(entitiesContainer, null);
@@ -306,12 +292,6 @@ public TimelinePutResponse putEntities(
   @Override
   public void putDomain(TimelineDomain domain) throws IOException,
       YarnException {
-    if (!isEnabled) {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Nothing will be put because timeline service is not enabled");
-      }
-      return;
-    }
     doPosting(domain, "domain");
   }
 
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineClient.java
index fe66e745cb3..3b4d6bdec64 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineClient.java
@@ -119,41 +119,6 @@ public void testPostEntitiesConnectionRefused() throws Exception {
     }
   }
 
-  @Test
-  public void testPostEntitiesTimelineServiceNotEnabled() throws Exception {
-    YarnConfiguration conf = new YarnConfiguration();
-    conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, false);
-    TimelineClientImpl client = createTimelineClient(conf);
-    mockEntityClientResponse(
-        client, ClientResponse.Status.INTERNAL_SERVER_ERROR, false, false);
-    try {
-      TimelinePutResponse response = client.putEntities(generateEntity());
-      Assert.assertEquals(0, response.getErrors().size());
-    } catch (YarnException e) {
-      Assert.fail(
-          "putEntities should already return before throwing the exception");
-    }
-  }
-
-  @Test
-  public void testPostEntitiesTimelineServiceDefaultNotEnabled()
-      throws Exception {
-    YarnConfiguration conf = new YarnConfiguration();
-    // Unset the timeline service's enabled properties.
-    // Make sure default value is pickup up
-    conf.unset(YarnConfiguration.TIMELINE_SERVICE_ENABLED);
-    TimelineClientImpl client = createTimelineClient(conf);
-    mockEntityClientResponse(client, ClientResponse.Status.INTERNAL_SERVER_ERROR,
-        false, false);
-    try {
-      TimelinePutResponse response = client.putEntities(generateEntity());
-      Assert.assertEquals(0, response.getErrors().size());
-    } catch (YarnException e) {
-      Assert
-          .fail("putEntities should already return before throwing the exception");
-    }
-  }
-
   @Test
   public void testPutDomain() throws Exception {
     mockDomainClientResponse(client, ClientResponse.Status.OK, false);