YARN-8073 TimelineClientImpl doesn't honor yarn.timeline-service.versions configuration. Contributed by Rohith Sharma K S
This commit is contained in:
parent
3087e89135
commit
345e7624d5
|
@ -269,7 +269,7 @@ public class JobHistoryEventHandler extends AbstractService
|
|||
LOG.info("Emitting job history data to the timeline service is enabled");
|
||||
if (YarnConfiguration.timelineServiceEnabled(conf)) {
|
||||
boolean timelineServiceV2Enabled =
|
||||
((int) YarnConfiguration.getTimelineServiceVersion(conf) == 2);
|
||||
YarnConfiguration.timelineServiceV2Enabled(conf);
|
||||
if(timelineServiceV2Enabled) {
|
||||
timelineV2Client =
|
||||
((MRAppMaster.RunningAppContext)context).getTimelineV2Client();
|
||||
|
|
|
@ -3796,6 +3796,27 @@ public class YarnConfiguration extends Configuration {
|
|||
return enabled;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns whether the timeline service v.1,5 is enabled via configuration.
|
||||
*
|
||||
* @param conf the configuration
|
||||
* @return whether the timeline service v.1.5 is enabled. V.1.5 refers to a
|
||||
* version equal to 1.5.
|
||||
*/
|
||||
public static boolean timelineServiceV15Enabled(Configuration conf) {
|
||||
boolean enabled = false;
|
||||
if (timelineServiceEnabled(conf)) {
|
||||
Collection<Float> versions = getTimelineServiceVersions(conf);
|
||||
for (Float version : versions) {
|
||||
if (Float.compare(version, 1.5f) == 0) {
|
||||
enabled = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
return enabled;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns all the active timeline service versions. It does not check
|
||||
* whether the timeline service itself is enabled.
|
||||
|
|
|
@ -82,7 +82,7 @@ public class TimelineClientImpl extends TimelineClient {
|
|||
@VisibleForTesting
|
||||
protected String doAsUser;
|
||||
|
||||
private float timelineServiceVersion;
|
||||
private boolean timelineServiceV15Enabled;
|
||||
private TimelineWriter timelineWriter;
|
||||
|
||||
private String timelineServiceAddress;
|
||||
|
@ -96,15 +96,15 @@ public class TimelineClientImpl extends TimelineClient {
|
|||
}
|
||||
|
||||
protected void serviceInit(Configuration conf) throws Exception {
|
||||
timelineServiceVersion =
|
||||
conf.getFloat(YarnConfiguration.TIMELINE_SERVICE_VERSION,
|
||||
YarnConfiguration.DEFAULT_TIMELINE_SERVICE_VERSION);
|
||||
if (!YarnConfiguration.timelineServiceV1Enabled(conf)) {
|
||||
throw new IOException("Timeline V1 client is not properly configured. "
|
||||
+ "Either timeline service is not enabled or version is not set to"
|
||||
+ " 1.x");
|
||||
}
|
||||
LOG.info("Timeline service address: " + getTimelineServiceAddress());
|
||||
|
||||
timelineServiceV15Enabled =
|
||||
YarnConfiguration.timelineServiceV15Enabled(conf);
|
||||
|
||||
UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
|
||||
UserGroupInformation realUgi = ugi.getRealUser();
|
||||
if (realUgi != null) {
|
||||
|
@ -126,6 +126,7 @@ public class TimelineClientImpl extends TimelineClient {
|
|||
conf.get(YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS,
|
||||
YarnConfiguration.DEFAULT_TIMELINE_SERVICE_WEBAPP_ADDRESS);
|
||||
}
|
||||
LOG.info("Timeline service address: " + getTimelineServiceAddress());
|
||||
super.serviceInit(conf);
|
||||
}
|
||||
|
||||
|
@ -147,7 +148,7 @@ public class TimelineClientImpl extends TimelineClient {
|
|||
protected TimelineWriter createTimelineWriter(Configuration conf,
|
||||
UserGroupInformation ugi, Client webClient, URI uri)
|
||||
throws IOException {
|
||||
if (Float.compare(this.timelineServiceVersion, 1.5f) == 0) {
|
||||
if (timelineServiceV15Enabled) {
|
||||
return new FileSystemTimelineWriter(
|
||||
conf, ugi, webClient, uri);
|
||||
} else {
|
||||
|
@ -406,10 +407,9 @@ public class TimelineClientImpl extends TimelineClient {
|
|||
public TimelinePutResponse putEntities(ApplicationAttemptId appAttemptId,
|
||||
TimelineEntityGroupId groupId, TimelineEntity... entities)
|
||||
throws IOException, YarnException {
|
||||
if (Float.compare(this.timelineServiceVersion, 1.5f) != 0) {
|
||||
if (!timelineServiceV15Enabled) {
|
||||
throw new YarnException(
|
||||
"This API is not supported under current Timeline Service Version: "
|
||||
+ timelineServiceVersion);
|
||||
"This API is not supported under current Timeline Service Version:");
|
||||
}
|
||||
|
||||
return timelineWriter.putEntities(appAttemptId, groupId, entities);
|
||||
|
@ -418,10 +418,9 @@ public class TimelineClientImpl extends TimelineClient {
|
|||
@Override
|
||||
public void putDomain(ApplicationAttemptId appAttemptId,
|
||||
TimelineDomain domain) throws IOException, YarnException {
|
||||
if (Float.compare(this.timelineServiceVersion, 1.5f) != 0) {
|
||||
if (!timelineServiceV15Enabled) {
|
||||
throw new YarnException(
|
||||
"This API is not supported under current Timeline Service Version: "
|
||||
+ timelineServiceVersion);
|
||||
"This API is not supported under current Timeline Service Version:");
|
||||
}
|
||||
timelineWriter.putDomain(appAttemptId, domain);
|
||||
}
|
||||
|
|
|
@ -126,8 +126,7 @@ public class TimelineUtils {
|
|||
* version equal to 1.5.
|
||||
*/
|
||||
public static boolean timelineServiceV1_5Enabled(Configuration conf) {
|
||||
return timelineServiceEnabled(conf) &&
|
||||
Math.abs(getTimelineServiceVersion(conf) - 1.5) < 0.00001;
|
||||
return YarnConfiguration.timelineServiceV15Enabled(conf);
|
||||
}
|
||||
|
||||
public static TimelineAbout createTimelineAbout(String about) {
|
||||
|
|
|
@ -31,6 +31,7 @@ import java.util.*;
|
|||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.fs.FileContext;
|
||||
import org.apache.hadoop.fs.Path;
|
||||
import org.apache.hadoop.yarn.api.records.*;
|
||||
|
@ -266,6 +267,31 @@ public class TestCombinedSystemMetricsPublisher {
|
|||
runTest(false, false);
|
||||
}
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testTimelineServiceConfiguration()
|
||||
throws Exception {
|
||||
Configuration config = new Configuration(false);
|
||||
config.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
|
||||
config.set(YarnConfiguration.TIMELINE_SERVICE_VERSIONS, "2.0,1.5");
|
||||
config.set(YarnConfiguration.TIMELINE_SERVICE_VERSION, "2.0");
|
||||
|
||||
Assert.assertTrue(YarnConfiguration.timelineServiceV2Enabled(config));
|
||||
Assert.assertTrue(YarnConfiguration.timelineServiceV15Enabled(config));
|
||||
Assert.assertTrue(YarnConfiguration.timelineServiceV1Enabled(config));
|
||||
|
||||
config.set(YarnConfiguration.TIMELINE_SERVICE_VERSIONS, "2.0,1");
|
||||
config.set(YarnConfiguration.TIMELINE_SERVICE_VERSION, "1.5");
|
||||
Assert.assertTrue(YarnConfiguration.timelineServiceV2Enabled(config));
|
||||
Assert.assertFalse(YarnConfiguration.timelineServiceV15Enabled(config));
|
||||
Assert.assertTrue(YarnConfiguration.timelineServiceV1Enabled(config));
|
||||
|
||||
config.set(YarnConfiguration.TIMELINE_SERVICE_VERSIONS, "2.0");
|
||||
config.set(YarnConfiguration.TIMELINE_SERVICE_VERSION, "1.5");
|
||||
Assert.assertTrue(YarnConfiguration.timelineServiceV2Enabled(config));
|
||||
Assert.assertFalse(YarnConfiguration.timelineServiceV15Enabled(config));
|
||||
Assert.assertFalse(YarnConfiguration.timelineServiceV1Enabled(config));
|
||||
}
|
||||
|
||||
private void publishEvents(boolean v1Enabled, boolean v2Enabled) {
|
||||
long timestamp = (v1Enabled) ? 1 : 2;
|
||||
int id = (v2Enabled) ? 3 : 4;
|
||||
|
|
|
@ -235,7 +235,7 @@ public class TimelineReaderServer extends CompositeService {
|
|||
public static void main(String[] args) {
|
||||
Configuration conf = new YarnConfiguration();
|
||||
conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
|
||||
conf.setFloat(YarnConfiguration.TIMELINE_SERVICE_VERSION, 2.0f);
|
||||
conf.setFloat(YarnConfiguration.TIMELINE_SERVICE_VERSIONS, 2.0f);
|
||||
TimelineReaderServer server = startTimelineReaderServer(args, conf);
|
||||
server.join();
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue