diff --git a/Tethering/Android.bp b/Tethering/Android.bp
index d04660d..498a1fe 100644
--- a/Tethering/Android.bp
+++ b/Tethering/Android.bp
@@ -54,6 +54,7 @@
         "src/**/*.java",
         ":framework-connectivity-shared-srcs",
         ":services-tethering-shared-srcs",
+        ":statslog-connectivity-java-gen",
         ":statslog-tethering-java-gen",
     ],
     static_libs: [
diff --git a/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java b/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java
index 5c853f4..89e06da 100644
--- a/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java
+++ b/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java
@@ -33,10 +33,12 @@
 import static com.android.networkstack.tethering.BpfUtils.DOWNSTREAM;
 import static com.android.networkstack.tethering.BpfUtils.UPSTREAM;
 import static com.android.networkstack.tethering.TetheringConfiguration.DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS;
+import static com.android.networkstack.tethering.TetheringConfiguration.TETHER_ACTIVE_SESSIONS_METRICS;
 import static com.android.networkstack.tethering.UpstreamNetworkState.isVcnInterface;
 import static com.android.networkstack.tethering.util.TetheringUtils.getTetheringJniLibraryName;
 
 import android.app.usage.NetworkStatsManager;
+import android.content.Context;
 import android.net.INetd;
 import android.net.IpPrefix;
 import android.net.LinkProperties;
@@ -65,6 +67,7 @@
 import com.android.net.module.util.BpfDump;
 import com.android.net.module.util.BpfMap;
 import com.android.net.module.util.CollectionUtils;
+import com.android.net.module.util.DeviceConfigUtils;
 import com.android.net.module.util.IBpfMap;
 import com.android.net.module.util.InterfaceParams;
 import com.android.net.module.util.NetworkStackConstants;
@@ -84,6 +87,7 @@
 import com.android.net.module.util.netlink.NetlinkUtils;
 import com.android.networkstack.tethering.apishim.common.BpfCoordinatorShim;
 import com.android.networkstack.tethering.util.TetheringUtils.ForwardedStats;
+import com.android.server.ConnectivityStatsLog;
 
 import java.io.IOException;
 import java.net.Inet4Address;
@@ -148,6 +152,13 @@
 
     @VisibleForTesting
     static final int CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS = 60_000;
+    // The interval is set to 5 minutes to strike a balance between minimizing
+    // the amount of metrics data uploaded and providing sufficient resolution
+    // to track changes in forwarding rules. This choice considers the minimum
+    // push metrics sampling interval of 5 minutes and the 3-minute timeout
+    // for forwarding rules.
+    @VisibleForTesting
+    static final int CONNTRACK_METRICS_UPDATE_INTERVAL_MS = 300_000;
     @VisibleForTesting
     static final int NF_CONNTRACK_TCP_TIMEOUT_ESTABLISHED = 432_000;
     @VisibleForTesting
@@ -314,12 +325,23 @@
         scheduleConntrackTimeoutUpdate();
     };
 
+    private final boolean mSupportActiveSessionsMetrics;
+
+    // Runnable that used by scheduling next refreshing of conntrack metrics sampling.
+    private final Runnable mScheduledConntrackMetricsSampling = () -> {
+        uploadConntrackMetricsSample();
+        scheduleConntrackMetricsSampling();
+    };
+
     // TODO: add BpfMap<TetherDownstream64Key, TetherDownstream64Value> retrieving function.
     @VisibleForTesting
     public abstract static class Dependencies {
         /** Get handler. */
         @NonNull public abstract Handler getHandler();
 
+        /** Get context. */
+        @NonNull public abstract Context getContext();
+
         /** Get netd. */
         @NonNull public abstract INetd getNetd();
 
@@ -472,6 +494,19 @@
                 return null;
             }
         }
+
+        /** Send a TetheringActiveSessionsReported event. */
+        public void sendTetheringActiveSessionsReported(int lastMaxSessionCount) {
+            ConnectivityStatsLog.write(ConnectivityStatsLog.TETHERING_ACTIVE_SESSIONS_REPORTED,
+                    lastMaxSessionCount);
+        }
+
+        /**
+         * @see DeviceConfigUtils#isTetheringFeatureEnabled
+         */
+        public boolean isFeatureEnabled(Context context, String name) {
+            return DeviceConfigUtils.isTetheringFeatureEnabled(context, name);
+        }
     }
 
     @VisibleForTesting
@@ -508,32 +543,57 @@
         if (!mBpfCoordinatorShim.isInitialized()) {
             mLog.e("Bpf shim not initialized");
         }
+
+        // BPF IPv4 forwarding only supports on S+.
+        mSupportActiveSessionsMetrics = mDeps.isAtLeastS()
+                && mDeps.isFeatureEnabled(mDeps.getContext(), TETHER_ACTIVE_SESSIONS_METRICS);
     }
 
     /**
-     * Start BPF tethering offload stats and conntrack timeout polling.
+     * Start BPF tethering offload stats and conntrack polling.
      * Note that this can be only called on handler thread.
      */
-    private void startStatsAndConntrackTimeoutPolling() {
+    private void startStatsAndConntrackPolling() {
         schedulePollingStats();
         scheduleConntrackTimeoutUpdate();
+        if (mSupportActiveSessionsMetrics) {
+            scheduleConntrackMetricsSampling();
+        }
 
         mLog.i("Polling started.");
     }
 
     /**
-     * Stop BPF tethering offload stats and conntrack timeout polling.
+     * Stop BPF tethering offload stats and conntrack polling.
      * The data limit cleanup and the tether stats maps cleanup are not implemented here.
      * These cleanups rely on all IpServers calling #removeIpv6DownstreamRule. After the
      * last rule is removed from the upstream, #removeIpv6DownstreamRule does the cleanup
      * functionality.
      * Note that this can be only called on handler thread.
      */
-    private void stopStatsAndConntrackTimeoutPolling() {
+    private void stopStatsAndConntrackPolling() {
         // Stop scheduled polling conntrack timeout.
         if (mHandler.hasCallbacks(mScheduledConntrackTimeoutUpdate)) {
             mHandler.removeCallbacks(mScheduledConntrackTimeoutUpdate);
         }
+        // Stop scheduled polling conntrack metrics sampling and
+        // clear counters in case there is any counter unsync problem
+        // previously due to possible bpf failures.
+        // Normally this won't happen because all clients are cleared before
+        // reaching here. See IpServer.BaseServingState#exit().
+        if (mSupportActiveSessionsMetrics) {
+            if (mHandler.hasCallbacks(mScheduledConntrackMetricsSampling)) {
+                mHandler.removeCallbacks(mScheduledConntrackMetricsSampling);
+            }
+            final int currentCount = mBpfConntrackEventConsumer.getCurrentConnectionCount();
+            if (currentCount != 0) {
+                Log.wtf(TAG, "Unexpected CurrentConnectionCount: " + currentCount);
+            }
+            // Avoid sending metrics when tethering is about to close.
+            // This leads to a missing final sample before disconnect
+            // but avoids possibly duplicating the last metric in the upload.
+            mBpfConntrackEventConsumer.clearConnectionCounters();
+        }
         // Stop scheduled polling stats and poll the latest stats from BPF maps.
         if (mHandler.hasCallbacks(mScheduledPollingStats)) {
             mHandler.removeCallbacks(mScheduledPollingStats);
@@ -867,7 +927,7 @@
 
         // Start monitoring and polling when the first IpServer is added.
         if (mServedIpServers.isEmpty()) {
-            startStatsAndConntrackTimeoutPolling();
+            startStatsAndConntrackPolling();
             startConntrackMonitoring();
             mIpNeighborMonitor.start();
             mLog.i("Neighbor monitoring started.");
@@ -890,7 +950,7 @@
 
         // Stop monitoring and polling when the last IpServer is removed.
         if (mServedIpServers.isEmpty()) {
-            stopStatsAndConntrackTimeoutPolling();
+            stopStatsAndConntrackPolling();
             stopConntrackMonitoring();
             mIpNeighborMonitor.stop();
             mLog.i("Neighbor monitoring stopped.");
@@ -1031,6 +1091,10 @@
         for (final Tether4Key k : deleteDownstreamRuleKeys) {
             mBpfCoordinatorShim.tetherOffloadRuleRemove(DOWNSTREAM, k);
         }
+        if (mSupportActiveSessionsMetrics) {
+            mBpfConntrackEventConsumer.decreaseCurrentConnectionCount(
+                    deleteUpstreamRuleKeys.size());
+        }
 
         // Cleanup each upstream interface by a set which avoids duplicated work on the same
         // upstream interface. Cleaning up the same interface twice (or more) here may raise
@@ -1300,6 +1364,13 @@
         pw.increaseIndent();
         dumpCounters(pw);
         pw.decreaseIndent();
+
+        pw.println();
+        pw.println("mSupportActiveSessionsMetrics: " + mSupportActiveSessionsMetrics);
+        pw.println("getLastMaxConnectionCount: "
+                + mBpfConntrackEventConsumer.getLastMaxConnectionCount());
+        pw.println("getCurrentConnectionCount: "
+                + mBpfConntrackEventConsumer.getCurrentConnectionCount());
     }
 
     private void dumpStats(@NonNull IndentingPrintWriter pw) {
@@ -1991,6 +2062,21 @@
     // while TCP status is established.
     @VisibleForTesting
     class BpfConntrackEventConsumer implements ConntrackEventConsumer {
+        /**
+         * Tracks the current number of tethering connections and the maximum
+         * observed since the last metrics collection. Used to provide insights
+         * into the distribution of active tethering sessions for metrics reporting.
+
+         * These variables are accessed on the handler thread, which includes:
+         *  1. ConntrackEvents signaling the addition or removal of an IPv4 rule.
+         *  2. ConntrackEvents indicating the removal of a tethering client,
+         *     triggering the removal of associated rules.
+         *  3. Removal of the last IpServer, which resets counters to handle
+         *     potential synchronization issues.
+         */
+        private int mLastMaxConnectionCount = 0;
+        private int mCurrentConnectionCount = 0;
+
         // The upstream4 and downstream4 rules are built as the following tables. Only raw ip
         // upstream interface is supported. Note that the field "lastUsed" is only updated by
         // BPF program which records the last used time for a given rule.
@@ -2124,6 +2210,10 @@
                     return;
                 }
 
+                if (mSupportActiveSessionsMetrics) {
+                    decreaseCurrentConnectionCount(1);
+                }
+
                 maybeClearLimit(upstreamIndex);
                 return;
             }
@@ -2136,8 +2226,50 @@
 
             maybeAddDevMap(upstreamIndex, tetherClient.downstreamIfindex);
             maybeSetLimit(upstreamIndex);
-            mBpfCoordinatorShim.tetherOffloadRuleAdd(UPSTREAM, upstream4Key, upstream4Value);
-            mBpfCoordinatorShim.tetherOffloadRuleAdd(DOWNSTREAM, downstream4Key, downstream4Value);
+
+            final boolean addedUpstream = mBpfCoordinatorShim.tetherOffloadRuleAdd(
+                    UPSTREAM, upstream4Key, upstream4Value);
+            final boolean addedDownstream = mBpfCoordinatorShim.tetherOffloadRuleAdd(
+                    DOWNSTREAM, downstream4Key, downstream4Value);
+            if (addedUpstream != addedDownstream) {
+                Log.wtf(TAG, "The bidirectional rules should be added concurrently ("
+                        + "upstream: " + addedUpstream
+                        + ", downstream: " + addedDownstream + ")");
+                return;
+            }
+            if (mSupportActiveSessionsMetrics && addedUpstream && addedDownstream) {
+                mCurrentConnectionCount++;
+                mLastMaxConnectionCount = Math.max(mCurrentConnectionCount,
+                        mLastMaxConnectionCount);
+            }
+        }
+
+        public int getLastMaxConnectionAndResetToCurrent() {
+            final int ret = mLastMaxConnectionCount;
+            mLastMaxConnectionCount = mCurrentConnectionCount;
+            return ret;
+        }
+
+        /** For dumping current state only. */
+        public int getLastMaxConnectionCount() {
+            return mLastMaxConnectionCount;
+        }
+
+        public int getCurrentConnectionCount() {
+            return mCurrentConnectionCount;
+        }
+
+        public void decreaseCurrentConnectionCount(int count) {
+            mCurrentConnectionCount -= count;
+            if (mCurrentConnectionCount < 0) {
+                Log.wtf(TAG, "Unexpected mCurrentConnectionCount: "
+                        + mCurrentConnectionCount);
+            }
+        }
+
+        public void clearConnectionCounters() {
+            mCurrentConnectionCount = 0;
+            mLastMaxConnectionCount = 0;
         }
     }
 
@@ -2477,6 +2609,11 @@
         });
     }
 
+    private void uploadConntrackMetricsSample() {
+        mDeps.sendTetheringActiveSessionsReported(
+                mBpfConntrackEventConsumer.getLastMaxConnectionAndResetToCurrent());
+    }
+
     private void schedulePollingStats() {
         if (mHandler.hasCallbacks(mScheduledPollingStats)) {
             mHandler.removeCallbacks(mScheduledPollingStats);
@@ -2494,6 +2631,15 @@
                 CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS);
     }
 
+    private void scheduleConntrackMetricsSampling() {
+        if (mHandler.hasCallbacks(mScheduledConntrackMetricsSampling)) {
+            mHandler.removeCallbacks(mScheduledConntrackMetricsSampling);
+        }
+
+        mHandler.postDelayed(mScheduledConntrackMetricsSampling,
+                CONNTRACK_METRICS_UPDATE_INTERVAL_MS);
+    }
+
     // Return IPv6 downstream forwarding rule map. This is used for testing only.
     // Note that this can be only called on handler thread.
     @NonNull
diff --git a/Tethering/src/com/android/networkstack/tethering/Tethering.java b/Tethering/src/com/android/networkstack/tethering/Tethering.java
index 13f4f2a..1938a08 100644
--- a/Tethering/src/com/android/networkstack/tethering/Tethering.java
+++ b/Tethering/src/com/android/networkstack/tethering/Tethering.java
@@ -374,6 +374,11 @@
                     }
 
                     @NonNull
+                    public Context getContext() {
+                        return mContext;
+                    }
+
+                    @NonNull
                     public INetd getNetd() {
                         return mNetd;
                     }
diff --git a/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java b/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
index 298940e..c9817c9 100644
--- a/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
+++ b/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
@@ -144,6 +144,12 @@
     /** A flag for using synchronous or asynchronous state machine. */
     public static boolean USE_SYNC_SM = false;
 
+    /**
+     * A feature flag to control whether the active sessions metrics should be enabled.
+     * Disabled by default.
+     */
+    public static final String TETHER_ACTIVE_SESSIONS_METRICS = "tether_active_sessions_metrics";
+
     public final String[] tetherableUsbRegexs;
     public final String[] tetherableWifiRegexs;
     public final String[] tetherableWigigRegexs;
diff --git a/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java b/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java
index e54a7e0..5d22977 100644
--- a/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java
+++ b/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java
@@ -48,6 +48,7 @@
 import static com.android.net.module.util.netlink.StructNdMsg.NUD_FAILED;
 import static com.android.net.module.util.netlink.StructNdMsg.NUD_REACHABLE;
 import static com.android.net.module.util.netlink.StructNdMsg.NUD_STALE;
+import static com.android.networkstack.tethering.BpfCoordinator.CONNTRACK_METRICS_UPDATE_INTERVAL_MS;
 import static com.android.networkstack.tethering.BpfCoordinator.CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS;
 import static com.android.networkstack.tethering.BpfCoordinator.INVALID_MTU;
 import static com.android.networkstack.tethering.BpfCoordinator.NF_CONNTRACK_TCP_TIMEOUT_ESTABLISHED;
@@ -60,6 +61,7 @@
 import static com.android.networkstack.tethering.BpfUtils.DOWNSTREAM;
 import static com.android.networkstack.tethering.BpfUtils.UPSTREAM;
 import static com.android.networkstack.tethering.TetheringConfiguration.DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS;
+import static com.android.networkstack.tethering.TetheringConfiguration.TETHER_ACTIVE_SESSIONS_METRICS;
 import static com.android.testutils.MiscAsserts.assertSameElements;
 
 import static org.junit.Assert.assertArrayEquals;
@@ -87,6 +89,7 @@
 import static org.mockito.Mockito.when;
 
 import android.app.usage.NetworkStatsManager;
+import android.content.Context;
 import android.net.INetd;
 import android.net.InetAddresses;
 import android.net.IpPrefix;
@@ -140,6 +143,8 @@
 import com.android.testutils.DevSdkIgnoreRule.IgnoreUpTo;
 import com.android.testutils.TestBpfMap;
 import com.android.testutils.TestableNetworkStatsProviderCbBinder;
+import com.android.testutils.com.android.testutils.SetFeatureFlagsRule;
+import com.android.testutils.com.android.testutils.SetFeatureFlagsRule.FeatureFlag;
 
 import org.junit.Before;
 import org.junit.Rule;
@@ -171,6 +176,16 @@
     @Rule
     public final DevSdkIgnoreRule mIgnoreRule = new DevSdkIgnoreRule();
 
+    final HashMap<String, Boolean> mFeatureFlags = new HashMap<>();
+    // This will set feature flags from @FeatureFlag annotations
+    // into the map before setUp() runs.
+    @Rule
+    public final SetFeatureFlagsRule mSetFeatureFlagsRule =
+            new SetFeatureFlagsRule((name, enabled) -> {
+                mFeatureFlags.put(name, enabled);
+                return null;
+            }, (name) -> mFeatureFlags.getOrDefault(name, false));
+
     private static final boolean IPV4 = true;
     private static final boolean IPV6 = false;
 
@@ -406,6 +421,11 @@
                 return this;
             }
 
+            public Builder setPrivateAddress(Inet4Address privateAddr) {
+                mPrivateAddr = privateAddr;
+                return this;
+            }
+
             public Builder setRemotePort(int remotePort) {
                 mRemotePort = (short) remotePort;
                 return this;
@@ -429,6 +449,7 @@
 
     @Mock private NetworkStatsManager mStatsManager;
     @Mock private INetd mNetd;
+    @Mock private Context mMockContext;
     @Mock private IpServer mIpServer;
     @Mock private IpServer mIpServer2;
     @Mock private TetheringConfiguration mTetherConfig;
@@ -475,6 +496,11 @@
                     }
 
                     @NonNull
+                    public Context getContext() {
+                        return mMockContext;
+                    }
+
+                    @NonNull
                     public INetd getNetd() {
                         return mNetd;
                     }
@@ -546,6 +572,16 @@
                     public IBpfMap<S32, S32> getBpfErrorMap() {
                         return mBpfErrorMap;
                     }
+
+                    @Override
+                    public void sendTetheringActiveSessionsReported(int lastMaxSessionCount) {
+                        // No-op.
+                    }
+
+                    @Override
+                    public boolean isFeatureEnabled(Context context, String name) {
+                        return mFeatureFlags.getOrDefault(name, false);
+                    }
             });
 
     @Before public void setUp() {
@@ -1977,6 +2013,229 @@
         verify(mBpfDevMap, never()).updateEntry(any(), any());
     }
 
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS)
+    // BPF IPv4 forwarding only supports on S+.
+    @IgnoreUpTo(Build.VERSION_CODES.R)
+    @Test
+    public void testMaxConnectionCount_metricsEnabled() throws Exception {
+        doTestMaxConnectionCount(true);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS, enabled = false)
+    @Test
+    public void testMaxConnectionCount_metricsDisabled() throws Exception {
+        doTestMaxConnectionCount(false);
+    }
+
+    private void doTestMaxConnectionCount(final boolean supportActiveSessionsMetrics)
+            throws Exception {
+        final BpfCoordinator coordinator = makeBpfCoordinator();
+        initBpfCoordinatorForRule4(coordinator);
+        resetNetdAndBpfMaps();
+        assertEquals(0, mConsumer.getLastMaxConnectionAndResetToCurrent());
+
+        // Prepare add/delete rule events.
+        final ArrayList<ConntrackEvent> addRuleEvents = new ArrayList<>();
+        final ArrayList<ConntrackEvent> delRuleEvents = new ArrayList<>();
+        for (int i = 0; i < 5; i++) {
+            final ConntrackEvent addEvent = new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_NEW).setProto(IPPROTO_TCP).setRemotePort(i).build();
+            addRuleEvents.add(addEvent);
+            final ConntrackEvent delEvent = new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_DELETE).setProto(IPPROTO_TCP).setRemotePort(i).build();
+            delRuleEvents.add(delEvent);
+        }
+
+        // Add rules, verify counter increases.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(addRuleEvents.get(i));
+            assertConsumerCountersEquals(supportActiveSessionsMetrics ? i + 1 : 0);
+        }
+
+        // Add the same events again should not increase the counter because
+        // all events are already exist.
+        for (final ConntrackEvent event : addRuleEvents) {
+            mConsumer.accept(event);
+            assertConsumerCountersEquals(supportActiveSessionsMetrics ? 5 : 0);
+        }
+
+        // Verify removing non-existent items won't change the counters.
+        for (int i = 5; i < 8; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_DELETE).setProto(IPPROTO_TCP).setRemotePort(i).build());
+            assertConsumerCountersEquals(supportActiveSessionsMetrics ? 5 : 0);
+        }
+
+        // Verify remove the rules decrease the counter.
+        // Note the max counter returns the max, so it returns the count before deleting.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(delRuleEvents.get(i));
+            assertEquals(supportActiveSessionsMetrics ? 4 - i : 0,
+                    mConsumer.getCurrentConnectionCount());
+            assertEquals(supportActiveSessionsMetrics ? 5 - i : 0,
+                    mConsumer.getLastMaxConnectionCount());
+            assertEquals(supportActiveSessionsMetrics ? 5 - i : 0,
+                    mConsumer.getLastMaxConnectionAndResetToCurrent());
+        }
+
+        // Verify remove these rules again doesn't decrease the counter.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(delRuleEvents.get(i));
+            assertConsumerCountersEquals(0);
+        }
+    }
+
+    // Helper method to assert all counter values inside consumer.
+    private void assertConsumerCountersEquals(int expectedCount) {
+        assertEquals(expectedCount, mConsumer.getCurrentConnectionCount());
+        assertEquals(expectedCount, mConsumer.getLastMaxConnectionCount());
+        assertEquals(expectedCount, mConsumer.getLastMaxConnectionAndResetToCurrent());
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS)
+    // BPF IPv4 forwarding only supports on S+.
+    @IgnoreUpTo(Build.VERSION_CODES.R)
+    @Test
+    public void doTestMaxConnectionCount_removeClient_metricsEnabled() throws Exception {
+        doTestMaxConnectionCount_removeClient(true);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS, enabled = false)
+    @Test
+    public void doTestMaxConnectionCount_removeClient_metricsDisabled() throws Exception {
+        doTestMaxConnectionCount_removeClient(false);
+    }
+
+    private void doTestMaxConnectionCount_removeClient(final boolean supportActiveSessionsMetrics)
+            throws Exception {
+        final BpfCoordinator coordinator = makeBpfCoordinator();
+        initBpfCoordinatorForRule4(coordinator);
+        resetNetdAndBpfMaps();
+
+        // Add client information A and B on on the same downstream.
+        final ClientInfo clientA = new ClientInfo(DOWNSTREAM_IFINDEX, DOWNSTREAM_MAC,
+                PRIVATE_ADDR, MAC_A);
+        final ClientInfo clientB = new ClientInfo(DOWNSTREAM_IFINDEX, DOWNSTREAM_MAC,
+                PRIVATE_ADDR2, MAC_B);
+        coordinator.tetherOffloadClientAdd(mIpServer, clientA);
+        coordinator.tetherOffloadClientAdd(mIpServer, clientB);
+        assertClientInfoExists(mIpServer, clientA);
+        assertClientInfoExists(mIpServer, clientB);
+        assertEquals(0, mConsumer.getLastMaxConnectionAndResetToCurrent());
+
+        // Add some rules for both clients.
+        final int addr1RuleCount = 5;
+        final int addr2RuleCount = 3;
+
+        for (int i = 0; i < addr1RuleCount; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder()
+                    .setMsgType(IPCTNL_MSG_CT_NEW)
+                    .setProto(IPPROTO_TCP)
+                    .setRemotePort(i)
+                    .setPrivateAddress(PRIVATE_ADDR)
+                    .build());
+        }
+
+        for (int i = addr1RuleCount; i < addr1RuleCount + addr2RuleCount; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder()
+                    .setMsgType(IPCTNL_MSG_CT_NEW)
+                    .setProto(IPPROTO_TCP)
+                    .setRemotePort(i)
+                    .setPrivateAddress(PRIVATE_ADDR2)
+                    .build());
+        }
+
+        assertConsumerCountersEquals(
+                supportActiveSessionsMetrics ? addr1RuleCount + addr2RuleCount : 0);
+
+        // Remove 1 client. Since the 1st poll will return the LastMaxCounter and
+        // update it to the current, the max counter will be kept at 1st poll, while
+        // the current counter reflect the rule decreasing.
+        coordinator.tetherOffloadClientRemove(mIpServer, clientA);
+        assertEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0,
+                mConsumer.getCurrentConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr1RuleCount + addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr1RuleCount + addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionAndResetToCurrent());
+        // And all counters be updated at 2nd poll.
+        assertConsumerCountersEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0);
+
+        // Remove other client.
+        coordinator.tetherOffloadClientRemove(mIpServer, clientB);
+        assertEquals(0, mConsumer.getCurrentConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionAndResetToCurrent());
+        // All counters reach zero at 2nd poll.
+        assertConsumerCountersEquals(0);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS)
+    // BPF IPv4 forwarding only supports on S+.
+    @IgnoreUpTo(Build.VERSION_CODES.R)
+    @Test
+    public void testSendActiveSessionsReported_metricsEnabled() throws Exception {
+        doTestSendActiveSessionsReported(true);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS, enabled = false)
+    @Test
+    public void testSendActiveSessionsReported_metricsDisabled() throws Exception {
+        doTestSendActiveSessionsReported(false);
+    }
+
+    private void doTestSendActiveSessionsReported(final boolean supportActiveSessionsMetrics)
+            throws Exception {
+        final BpfCoordinator coordinator = makeBpfCoordinator();
+        initBpfCoordinatorForRule4(coordinator);
+        resetNetdAndBpfMaps();
+        assertConsumerCountersEquals(0);
+
+        // Prepare the counter value.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_NEW).setProto(IPPROTO_TCP).setRemotePort(i).build());
+        }
+
+        // Then delete some 3 rules, 2 rules remaining.
+        // The max count is 5 while current rules count is 2.
+        for (int i = 0; i < 3; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_DELETE).setProto(IPPROTO_TCP).setRemotePort(i).build());
+        }
+
+        // Verify the method is not invoked when timer is not expired.
+        waitForIdle();
+        verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+
+        // Verify metrics will be sent upon timer expiry.
+        mTestLooper.moveTimeForward(CONNTRACK_METRICS_UPDATE_INTERVAL_MS);
+        waitForIdle();
+        if (supportActiveSessionsMetrics) {
+            verify(mDeps).sendTetheringActiveSessionsReported(5);
+        } else {
+            verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+        }
+
+        // Verify next uploaded metrics will reflect the decreased rules count.
+        mTestLooper.moveTimeForward(CONNTRACK_METRICS_UPDATE_INTERVAL_MS);
+        waitForIdle();
+        if (supportActiveSessionsMetrics) {
+            verify(mDeps).sendTetheringActiveSessionsReported(2);
+        } else {
+            verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+        }
+
+        // Verify no metrics uploaded if polling stopped.
+        clearInvocations(mDeps);
+        coordinator.removeIpServer(mIpServer);
+        mTestLooper.moveTimeForward(CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS);
+        waitForIdle();
+        verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+    }
+
     private void setElapsedRealtimeNanos(long nanoSec) {
         mElapsedRealtimeNanos = nanoSec;
     }