| /* |
| * 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.samza.job.yarn |
| |
| import java.io.{BufferedReader, InputStreamReader} |
| import java.net.URL |
| |
| import org.apache.hadoop.yarn.util.ConverterUtils |
| import org.apache.samza.clustermanager.{ClusterBasedJobCoordinator, SamzaApplicationState} |
| import org.apache.samza.config.{Config, MapConfig} |
| import org.apache.samza.coordinator.JobModelManager |
| import org.apache.samza.coordinator.metadatastore.{CoordinatorStreamStore, CoordinatorStreamStoreTestUtil, NamespaceAwareCoordinatorStreamStore} |
| import org.apache.samza.coordinator.stream.messages.SetChangelogMapping |
| import org.apache.samza.coordinator.stream.MockCoordinatorStreamSystemFactory |
| import org.apache.samza.metrics._ |
| import org.apache.samza.storage.ChangelogStreamManager |
| import org.junit.Assert._ |
| import org.junit.Test |
| |
| import scala.collection.JavaConverters._ |
| |
| class TestSamzaYarnAppMasterService { |
| |
| @Test |
| def testAppMasterDashboardShouldStart { |
| MockCoordinatorStreamSystemFactory.enableMockConsumerCache() |
| val config = getDummyConfig |
| val jobModelManager = getTestJobModelManager(config) |
| val samzaState = new SamzaApplicationState(jobModelManager) |
| val registry = new MetricsRegistryMap() |
| |
| val state = new YarnAppState(-1, ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "testHost", 1, 1) |
| val service = new SamzaYarnAppMasterService(config, samzaState, state, registry, null) |
| |
| // start the dashboard |
| service.onInit |
| assertTrue(state.rpcUrl.getPort > 0) |
| assertTrue(state.trackingUrl.getPort > 0) |
| assertTrue(state.coordinatorUrl.getPort > 0) |
| |
| // check to see if it's running |
| val url = new URL(state.rpcUrl.toString + "am") |
| val is = url.openConnection().getInputStream() |
| val reader = new BufferedReader(new InputStreamReader(is)) |
| var line: String = null |
| |
| do { |
| line = reader.readLine() |
| } while (line != null) |
| |
| reader.close() |
| } |
| |
| /** |
| * This tests the rendering of the index.scaml file containing some Scala code. The objective |
| * is to ensure that the rendered scala code builds correctly |
| */ |
| @Test |
| def testAppMasterDashboardWebServiceShouldStart { |
| // Create some dummy config |
| val config = getDummyConfig |
| val jobModelManager = getTestJobModelManager(config) |
| val samzaState = new SamzaApplicationState(jobModelManager) |
| val state = new YarnAppState(-1, ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "testHost", 1, 1) |
| val registry = new MetricsRegistryMap() |
| |
| val service = new SamzaYarnAppMasterService(config, samzaState, state, registry, null) |
| |
| // start the dashboard |
| service.onInit |
| assertTrue(state.rpcUrl.getPort > 0) |
| assertTrue(state.trackingUrl.getPort > 0) |
| |
| // Do a GET Request on the tracking port: This in turn will render index.scaml |
| val url = state.trackingUrl |
| val is = url.openConnection().getInputStream() |
| val reader = new BufferedReader(new InputStreamReader(is)) |
| var line: String = null |
| |
| do { |
| line = reader.readLine() |
| } while (line != null) |
| |
| reader.close |
| } |
| |
| private def getTestJobModelManager(config: Config) = { |
| val coordinatorStreamTestUtil: CoordinatorStreamStoreTestUtil = new CoordinatorStreamStoreTestUtil(config) |
| val coordinatorStreamStore: CoordinatorStreamStore = coordinatorStreamTestUtil.getCoordinatorStreamStore |
| val namespaceAwareCoordinatorStore: NamespaceAwareCoordinatorStreamStore = new NamespaceAwareCoordinatorStreamStore(coordinatorStreamStore, SetChangelogMapping.TYPE) |
| try { |
| val changelogPartitionManager = new ChangelogStreamManager(namespaceAwareCoordinatorStore) |
| val jobModelManager = JobModelManager(getDummyConfig, changelogPartitionManager.readPartitionMapping(), |
| coordinatorStreamStore, new MetricsRegistryMap()) |
| jobModelManager |
| } finally { |
| coordinatorStreamStore.close() |
| } |
| } |
| |
| private def getDummyConfig: Config = new MapConfig(Map[String, String]( |
| "job.name" -> "test", |
| "yarn.container.count" -> "1", |
| "systems.test-system.samza.factory" -> "org.apache.samza.job.yarn.MockSystemFactory", |
| "yarn.container.memory.mb" -> "512", |
| "yarn.package.path" -> "/foo", |
| "task.inputs" -> "test-system.test-stream", |
| "systems.test-system.samza.key.serde" -> "org.apache.samza.serializers.JsonSerde", |
| "systems.test-system.samza.msg.serde" -> "org.apache.samza.serializers.JsonSerde", |
| "yarn.container.retry.count" -> "1", |
| "yarn.container.retry.window.ms" -> "1999999999", |
| "job.coordinator.system" -> "coordinator", |
| "systems.coordinator.samza.factory" -> classOf[MockCoordinatorStreamSystemFactory].getCanonicalName).asJava) |
| } |