Skip to content

EN-62834: Bring back exlusive-lock-on-resync #413

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
rjmac wants to merge 1 commit into
base: main
Choose a base branch
from
Open

EN-62834: Bring back exlusive-lock-on-resync #413

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
17 changes: 15 additions & 2 deletions ...dinatorlib/src/main/scala/com/socrata/datacoordinator/secondary/PlaybackToSecondary.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import org.slf4j.LoggerFactory

object PlaybackToSecondary {
type SuperUniverse[CT, CV] = Universe[CT, CV] with Commitable with
SecondaryStoresConfigProvider with
SecondaryManifestProvider with
SecondaryMetricsProvider with
DatasetMapReaderProvider with
Expand Down Expand Up @@ -679,13 +680,25 @@ class PlaybackToSecondary[CT, CV](u: PlaybackToSecondary.SuperUniverse[CT, CV],
}
}

def dsInfoOf(datasetId: DatasetId) = {
u.secondaryStoresConfig.lookup(secondary.storeId) match {
case Some(config) if config.isFeedback =>
logger.info("Non-exclusive resync")
u.datasetMapReader.datasetInfo(datasetId, repeatableRead = true)
case _ =>
logger.info("exclusive resync")
u.datasetMapWriter.datasetInfo(datasetId, datasetLockTimeout, semiExclusive = true)
}
}

def resync(): Unit = {
val mostRecentlyUpdatedCopyInfo = retrying[Option[metadata.CopyInfo]]({
timingReport("resync", "dataset" -> datasetId) {
u.commit() // all updates must be committed before we can change the transaction isolation level
val r = u.datasetMapReader
r.datasetInfo(datasetId, repeatableRead = true) match {
// transaction isolation level is now set to REPEATABLE READ
val w = u.datasetMapWriter
dsInfoOf(datasetId) match {
// transaction isolation level is now set to REPEATABLE READ or we have a lock on the dataset
case Some(datasetInfo) =>
val allCopies = r.allCopies(datasetInfo).toSeq.sortBy(_.dataVersion)
val mostRecentCopy =
Expand Down
8 changes: 7 additions & 1 deletion coordinatorlib/src/main/scala/com/socrata/datacoordinator/secondary/SecondaryInfo.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,13 @@ package com.socrata.datacoordinator.secondary

import org.joda.time.DateTime

case class SecondaryConfigInfo(storeId: String, nextRunTime: DateTime, runIntervalSeconds: Int, groupName: String)
case class SecondaryConfigInfo(
storeId: String,
nextRunTime: DateTime,
runIntervalSeconds: Int,
groupName: String,
isFeedback: Boolean
)

trait SecondaryStoresConfig {
def lookup(storeId: String): Option[SecondaryConfigInfo]
Expand Down
5 changes: 3 additions & 2 deletions ...b/src/main/scala/com/socrata/datacoordinator/secondary/sql/SqlSecondaryStoresConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ class SqlSecondaryStoresConfig(conn: Connection, timingReport: TimingReport) ext

def lookup(storeId: String): Option[SecondaryConfigInfo] = {
val sql = """
SELECT store_id, next_run_time, interval_in_seconds, group_name
SELECT store_id, next_run_time, interval_in_seconds, group_name, is_feedback_secondary
FROM secondary_stores_config
WHERE store_id = ?""".stripMargin

Expand All @@ -26,7 +26,8 @@ class SqlSecondaryStoresConfig(conn: Connection, timingReport: TimingReport) ext
rs.getString("store_id"),
new DateTime(rs.getTimestamp("next_run_time").getTime),
rs.getInt("interval_in_seconds"),
Option(rs.getString("group_name")).getOrElse("")))
Option(rs.getString("group_name")).getOrElse(""),
rs.getBoolean("is_feedback_secondary")))
} else {
None
}
Expand Down