Serialize writes to each table
@@ -105,6 +105,7 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
}
private void addNodeMetrics(Collection<Pair<String, NodeMetricSnapshot>> snapshots, TableWriter writer) {
+ synchronized (nodeTable.writeLock) {
for (var snapshot : snapshots) {
Optional<Long> atMillis = nodeTable.adjustOrDiscard(snapshot.getSecond().at());
if (atMillis.isEmpty()) return;
@@ -122,6 +123,7 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
}
writer.commit();
}
+ }
@Override
public void addClusterMetrics(ApplicationId application, Map<ClusterSpec.Id, ClusterMetricSnapshot> snapshots) {
@@ -140,6 +142,7 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
}
private void addClusterMetrics(ApplicationId applicationId, Map<ClusterSpec.Id, ClusterMetricSnapshot> snapshots, TableWriter writer) {
+ synchronized (clusterTable.writeLock) {
for (var snapshot : snapshots.entrySet()) {
Optional<Long> atMillis = clusterTable.adjustOrDiscard(snapshot.getValue().at());
if (atMillis.isEmpty()) return;
@@ -153,6 +156,7 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
}
writer.commit();
}
+ }
@Override
public List<NodeTimeseries> getNodeTimeseries(Duration period, Set<String> hostnames) {
@@ -330,6 +334,7 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
/** A questDb table */
private class Table {
+ private final Object writeLock = new Object();
private final String name;
private final Clock clock;
private final File dir;
