@Override
    public void run() {
      ReplicationMetricList metricList = new ReplicationMetricList();
      try {
        LOG.debug("Updating metrics to DB");
        // get metrics
        LinkedList<ReplicationMetric> metrics = collector.getMetrics();
        //Move metrics to thrift list
        if (metrics.size() > 0) {
          LOG.debug("Converting metrics to thrift metrics {} ", metrics.size());
          int totalMetricsSize = metrics.size();
          List<ReplicationMetrics> replicationMetricsList = new ArrayList<>(totalMetricsSize);
          ObjectMapper mapper = new ObjectMapper();
          MessageEncoder encoder = MessageFactory.getDefaultInstanceForReplMetrics(conf);
          for (int index = 0; index < totalMetricsSize; index++) {
            ReplicationMetric metric = metrics.removeFirst();
            ReplicationMetrics persistentMetric = new ReplicationMetrics();
            persistentMetric.setDumpExecutionId(metric.getDumpExecutionId());
            persistentMetric.setScheduledExecutionId(metric.getScheduledExecutionId());
            persistentMetric.setPolicy(metric.getPolicy());
            persistentMetric.setProgress(updateRMProgressIfLimitExceeds(metric.getProgress(), encoder));
            persistentMetric.setMetadata(mapper.writeValueAsString(metric.getMetadata()));
            persistentMetric.setMessageFormat(encoder.getMessageFormat());
            LOG.debug("Metric to be persisted {} ", persistentMetric);
            replicationMetricsList.add(persistentMetric);
          }
          metricList.setReplicationMetricList(replicationMetricsList);
          // write metrics and retry if fails
          Retryable retryable = Retryable.builder()
            .withHiveConf(conf)
            .withRetryOnException(Exception.class).build();
          retryable.executeCallable((Callable<Void>) () -> {
            if (metricList.getReplicationMetricListSize() > 0) {
              LOG.debug("Persisting metrics to DB {} ", metricList.getReplicationMetricListSize());
              Hive.get(conf).getMSC().addReplicationMetrics(metricList);
            }
            return null;
          });
        } else {
          LOG.debug("No Metrics to Update ");
        }
      } catch (Exception e) {
        LOG.error("Metrics are not getting persisted", e);
      }
    }
MetricSink.
Scheduled thread to poll from Metric Collector and persists to DB
ql/src/java/org/apache/hadoop/hive/ql/parse/repl/metric/MetricSink.java