Fix the persistent session timeout mechanism bug. (#7347)

CHANGES.md

@@ -112,6 +112,7 @@ Release Notes.
 * Enhance persistent session timeout mechanism. Because the enhanced session could cache the metadata metrics forever,
   new timeout mechanism is designed for avoiding this specific case.
 * Fix Kafka transport topics are created duplicated with and without namespace issue
+* Fix the persistent session timeout mechanism bug.
 * Fix possible version_conflict_engine_exception in bulk execution.
 * Fix PrometheusMetricConverter may throw an `IllegalArgumentException` when convert metrics to SampleFamily
 * Filtering NaN value samples when build SampleFamily  

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java

@@ -275,7 +275,8 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics> {
                                // Mostly all updatable metadata level metrics are required to do this check.
 
                                if (metricsDAO.isExpiredCache(model, cachedValue, currentTimeMillis, metricsDataTTL)) {
-                                   // The expired metrics should be tagged `not in cache` directly.
+                                   // The expired metrics should be removed from the context and tagged `not in cache` directly.
+                                   context.remove(m);
                                    return true;
                                }
                            }

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java

@@ -135,7 +135,7 @@ public class MetricsEsDAO extends EsDAO implements IMetricsDAO {
             return false;
         }
         final long deadline = Long.parseLong(new DateTime(currentTimeMillis).plusDays(-ttl).toString("yyyyMMdd"));
-        final long timeBucket = TimeBucket.getTimeBucket(cachedValue.getTimeBucket(), DownSampling.Day);
+        final long timeBucket = TimeBucket.getTimeBucket(metricTimestamp, DownSampling.Day);
         // If time bucket is earlier or equals(mostly) the deadline, then the cached metric is expired.
         if (timeBucket <= deadline) {
             return true;