Fix InstanceTraffic.combine overwriting last_ping with an older minute

docs/en/changes/changes.md

@@ -48,6 +48,7 @@
 * Fix the Elasticsearch Zipkin trace search listing the daily indices of every day since 1970 when the query starts at or before the epoch, as a `lookback` equal to `endTs` makes it. With the default day step of 1 that is over 20,000 index names, which can exceed the request line Elasticsearch accepts. The search now uses the whole `zipkin_span` index family in that case, as its time filter already did. Queries with a usual `lookback` are not affected.
 * Fix Elasticsearch reads of a metric a peer OAP reinstalls without a schema change, such as a bundled MAL rule restored after the runtime rule overriding it is deleted. The peer kept the index it had registered for the runtime rule. This only matters with `logicSharding` enabled and when the two rules use different functions, since the function decides the index there; with the default merged indices every metric stays in `metrics-all`.
 * Build the dated Elasticsearch index names in one place, `IndexNames`, with the day step decided once per physical table from its models, replacing `TimeSeriesUtils` and `TimeRangeIndexNameGenerator`. Apart from the fixes above, index names are unchanged.
+* Fix a service instance dropping out of the instance list queried with a time range when a record carrying an older minute updates it, such as the segment of a request that started minutes earlier arriving after the instance's heartbeat. `InstanceTraffic` took the `last_ping` of the last record written instead of keeping the latest one, as `EndpointTraffic` now does.
 
 #### UI
 * Add a Virtual GenAI evaluation-record page and evaluation-score chart in Horizon UI, so operators can inspect evaluation result, level, reason, judge model, timestamp, trace linkage, and the `gen_ai_model_evaluation_score_ppm` trend for evaluated records.

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/InstanceTraffic.java

@@ -82,7 +82,8 @@ public class InstanceTraffic extends Metrics {
     @Override
     public boolean combine(final Metrics metrics) {
         final InstanceTraffic instanceTraffic = (InstanceTraffic) metrics;
-        this.lastPingTimestamp = instanceTraffic.getLastPingTimestamp();
+        // Several sources feed one instance, and a late one may carry an older minute.
+        this.lastPingTimestamp = Math.max(this.lastPingTimestamp, instanceTraffic.getLastPingTimestamp());
         if (instanceTraffic.getProperties() != null && instanceTraffic.getProperties().size() > 0) {
             if (this.properties == null) {
                 this.properties = new JsonObject();

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/InstanceTrafficTest.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.analysis.manual.instance;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class InstanceTrafficTest {
+    @Test
+    public void testCombineKeepsLatestLastPing() {
+        final InstanceTraffic stored = newTraffic(202003281425L);
+
+        stored.combine(newTraffic(202003281421L));
+        Assertions.assertEquals(202003281425L, stored.getLastPingTimestamp());
+
+        stored.combine(newTraffic(202003281430L));
+        Assertions.assertEquals(202003281430L, stored.getLastPingTimestamp());
+    }
+
+    private static InstanceTraffic newTraffic(final long lastPing) {
+        final InstanceTraffic traffic = new InstanceTraffic();
+        traffic.setTimeBucket(202003281420L);
+        traffic.setServiceId("mock_id");
+        traffic.setName("instance-123");
+        traffic.setLastPingTimestamp(lastPing);
+        return traffic;
+    }
+}