Skip to content

Updating the chronos and zk communication for leader election and task storage #900

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: stable
Choose a base branch
from
Open

Updating the chronos and zk communication for leader election and task storage #900

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e9069dd
upgraded dependencies and now building with jdk11.
tushar-os Sep 10, 2019
47d761c
minor dependency change.
tushar-os Sep 10, 2019
54abf00
updated the deprecated classes.
tushar-os Jan 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 36 additions & 16 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,12 @@
<assets.dir>${basedir}/src/main/resources/assets</assets.dir>

<!-- language versions -->
<java.abi>1.7</java.abi>
<scala.version>2.11.2</scala.version>
<scala.abi>2.11.2</scala.abi>
<java.abi>11</java.abi>
<scala.version>2.11.12</scala.version>
<scala.abi>2.11.12</scala.abi>

<!-- runtime deps versions -->
<akka.version>2.3.16</akka.version>
<akka.version>2.4.12</akka.version>
<cassandra-driver-core.version>3.1.2</cassandra-driver-core.version>
<chaos.version>0.6.7</chaos.version>
<commons-math3.version>3.2</commons-math3.version>
Expand All @@ -36,20 +36,20 @@
<raven.version>7.8.0</raven.version>

<!-- test deps versions -->
<akka-testkit.version>2.3.16</akka-testkit.version>
<junit.version>4.11</junit.version>
<mockito.version>1.9.5</mockito.version>
<specs2.version>3.6.4-20150927082714-7cac887</specs2.version>
<akka-testkit.version>2.4.12</akka-testkit.version>
<junit.version>4.12</junit.version>
<mockito.version>2.0.2-beta</mockito.version>
<specs2.version>3.8.6</specs2.version>

<!-- maven versions-->
<maven-antrun-plugin.version>1.7</maven-antrun-plugin.version>
<maven-clean-plugin.version>2.5</maven-clean-plugin.version>
<maven-compiler-plugin.version>3.1</maven-compiler-plugin.version>
<maven-javadoc-plugin.version>2.9</maven-javadoc-plugin.version>
<maven-jar-plugin.version>2.4</maven-jar-plugin.version>
<maven-shade-plugin.version>2.3</maven-shade-plugin.version>
<maven-surefire-plugins.version>2.14.1</maven-surefire-plugins.version>
<scala-maven-plugin.version>3.1.0</scala-maven-plugin.version>
<maven-antrun-plugin.version>1.8</maven-antrun-plugin.version>
<maven-clean-plugin.version>3.1.0</maven-clean-plugin.version>
<maven-compiler-plugin.version>3.8.0</maven-compiler-plugin.version>
<maven-javadoc-plugin.version>3.1.0</maven-javadoc-plugin.version>
<maven-jar-plugin.version>3.1.1</maven-jar-plugin.version>
<maven-shade-plugin.version>3.2.1</maven-shade-plugin.version>
<maven-surefire-plugins.version>3.0.0-M3</maven-surefire-plugins.version>
<scala-maven-plugin.version>4.0.1</scala-maven-plugin.version>
</properties>

<repositories>
Expand Down Expand Up @@ -238,6 +238,26 @@
<version>${specs2.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>
<version>2.2.11</version>
</dependency>
<dependency>
<groupId>com.sun.xml.bind</groupId>
<artifactId>jaxb-core</artifactId>
<version>2.2.11</version>
</dependency>
<dependency>
<groupId>com.sun.xml.bind</groupId>
<artifactId>jaxb-impl</artifactId>
<version>2.2.11</version>
</dependency>
<dependency>
<groupId>javax.activation</groupId>
<artifactId>activation</artifactId>
<version>1.1.1</version>
</dependency>
<dependency>
<groupId>org.specs2</groupId>
<artifactId>specs2-junit_2.11</artifactId>
Expand Down
15 changes: 10 additions & 5 deletions src/main/scala/org/apache/mesos/chronos/scheduler/config/ZookeeperModule.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@ import mesosphere.mesos.util.FrameworkIdUtil
import org.apache.curator.framework.recipes.leader.LeaderLatch
import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
import org.apache.curator.retry.ExponentialBackoffRetry
import org.apache.curator.utils.EnsurePath
import org.apache.mesos.state.{State, ZooKeeperState}

/**
Expand Down Expand Up @@ -72,8 +71,11 @@ class ZookeeperModule(val config: SchedulerConfiguration with HttpConf)
@Singleton
@Provides
def provideStore(zk: CuratorFramework, state: State): PersistenceStore = {
val ensurePath: EnsurePath = new EnsurePath(config.zooKeeperStatePath)
ensurePath.ensure(zk.getZookeeperClient)
if (zk.checkExists().forPath(config.zooKeeperStatePath) != null) {
zk.checkExists().creatingParentContainersIfNeeded()
} else {
zk.create().creatingParentContainersIfNeeded()
}

new MesosStatePersistenceStore(zk, config, state)
}
Expand All @@ -88,8 +90,11 @@ class ZookeeperModule(val config: SchedulerConfiguration with HttpConf)
@Singleton
@Provides
def provideLeaderLatch(curator: CuratorFramework): LeaderLatch = {
val ensurePath: EnsurePath = new EnsurePath(config.zooKeeperCandidatePath)
ensurePath.ensure(curator.getZookeeperClient)
if (curator.checkExists().forPath(config.zooKeeperCandidatePath) != null) {
curator.checkExists().creatingParentContainersIfNeeded()
} else {
curator.create().creatingParentContainersIfNeeded()
}

val id = "%s:%d".format(config.hostname(), config.httpPort())
new LeaderLatch(curator, config.zooKeeperCandidatePath, id)
Expand Down
9 changes: 6 additions & 3 deletions src/test/scala/org/apache/mesos/chronos/scheduler/jobs/JobSchedulerElectionSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import org.apache.curator.framework.recipes.leader.LeaderLatch
import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
import org.apache.curator.retry.ExponentialBackoffRetry
import org.apache.curator.test.{InstanceSpec, TestingCluster}
import org.apache.curator.utils.{CloseableUtils, EnsurePath}
import org.apache.curator.utils.CloseableUtils
import org.joda.time.Period
import org.junit.Assert.{assertFalse, assertTrue}
import org.junit.Test
Expand Down Expand Up @@ -135,8 +135,11 @@ class JobSchedulerElectionSpec
jobsObserver = mock[JobsObserver.Observer]
)

val ensurePath: EnsurePath = new EnsurePath(leaderPath)
ensurePath.ensure(curator.getZookeeperClient)
if (curator.checkExists().forPath(leaderPath) != null) {
curator.checkExists().creatingParentContainersIfNeeded()
} else {
curator.create().creatingParentContainersIfNeeded()
}

(scheduler, curator, leaderLatch)
}
Expand Down
16 changes: 9 additions & 7 deletions src/test/scala/org/apache/mesos/chronos/scheduler/mesos/MesosOfferReviverActorSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,17 @@ package org.apache.mesos.chronos.scheduler.mesos

import akka.actor._
import akka.testkit.TestProbe
import com.codahale.metrics.{ Counter, MetricRegistry }
import com.codahale.metrics.{Counter, MetricRegistry}
import mesosphere.chaos.http.HttpConf
import mesosphere.mesos.util.FrameworkIdUtil
import org.apache.mesos.SchedulerDriver
import org.apache.mesos.chronos.scheduler.config.SchedulerConfiguration
import org.specs2.matcher.ThrownExpectations
import org.specs2.mock.Mockito
import org.specs2.mutable._
import scala.concurrent.duration.FiniteDuration
import org.specs2.matcher.ThrownExpectations

import scala.concurrent.Await
import scala.concurrent.duration.{FiniteDuration, _}

class MesosOfferReviverActorSpec extends SpecificationWithJUnit with Mockito {
"MesosOfferReviverActor" should {
Expand Down Expand Up @@ -39,8 +41,8 @@ class MesosOfferReviverActorSpec extends SpecificationWithJUnit with Mockito {
probe.watch(actorRef)
probe.expectMsgAnyClassOf(classOf[Terminated])

actorSystem.shutdown()
actorSystem.awaitTermination()
actorSystem.terminate()
Await.result(actorSystem.whenTerminated, Duration.Inf)

there was one(reviveOffersCounter).inc()
}
Expand Down Expand Up @@ -98,8 +100,8 @@ trait context extends BeforeAfter with Mockito with ThrownExpectations {
}

def after = {
actorSystem.shutdown()
actorSystem.awaitTermination()
actorSystem.terminate()
Await.result(actorSystem.whenTerminated, Duration.Inf)
}

def startActor(props: Props = MesosOfferReviverActor.props(conf, driverFactory, metrics)): ActorRef = {
Expand Down