Serialize calls to table.getWriter
QuestDb devs suspects they have a bug with concurrent getWriter invocations
@@ -92,22 +92,21 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
@Override
public void addNodeMetrics(Collection<Pair<String, NodeMetricSnapshot>> snapshots) {
- try (TableWriter writer = nodeTable.getWriter()) {
- addNodeMetrics(snapshots, writer);
+ try {
+ addNodeMetricsBody(snapshots);
}
catch (CairoException e) {
if (e.getMessage().contains("Cannot read offset")) {
// This error seems non-recoverable
nodeTable.repair(e);
- try (TableWriter writer = nodeTable.getWriter()) {
- addNodeMetrics(snapshots, writer);
- }
+ addNodeMetricsBody(snapshots);
}
}
}
- private void addNodeMetrics(Collection<Pair<String, NodeMetricSnapshot>> snapshots, TableWriter writer) {
+ private void addNodeMetricsBody(Collection<Pair<String, NodeMetricSnapshot>> snapshots) {
synchronized (nodeTable.writeLock) {
+ try (TableWriter writer = nodeTable.getWriter()) {
for (var snapshot : snapshots) {
Optional<Long> atMillis = nodeTable.adjustOrDiscard(snapshot.getSecond().at());
if (atMillis.isEmpty()) continue;
@@ -126,25 +125,25 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
writer.commit();
}
}
+ }
@Override
public void addClusterMetrics(ApplicationId application, Map<ClusterSpec.Id, ClusterMetricSnapshot> snapshots) {
- try (TableWriter writer = clusterTable.getWriter()) {
- addClusterMetrics(application, snapshots, writer);
+ try {
+ addClusterMetricsBody(application, snapshots);
}
catch (CairoException e) {
if (e.getMessage().contains("Cannot read offset")) {
// This error seems non-recoverable
clusterTable.repair(e);
- try (TableWriter writer = clusterTable.getWriter()) {
- addClusterMetrics(application, snapshots, writer);
- }
+ addClusterMetricsBody(application, snapshots);
}
}
}
- private void addClusterMetrics(ApplicationId applicationId, Map<ClusterSpec.Id, ClusterMetricSnapshot> snapshots, TableWriter writer) {
+ private void addClusterMetricsBody(ApplicationId applicationId, Map<ClusterSpec.Id, ClusterMetricSnapshot> snapshots) {
synchronized (clusterTable.writeLock) {
+ try (TableWriter writer = clusterTable.getWriter()) {
for (var snapshot : snapshots.entrySet()) {
Optional<Long> atMillis = clusterTable.adjustOrDiscard(snapshot.getValue().at());
if (atMillis.isEmpty()) continue;
@@ -159,6 +158,7 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
writer.commit();
}
}
+ }
@Override
public List<NodeTimeseries> getNodeTimeseries(Duration period, Set<String> hostnames) {
