diff --git a/Tethering/res/values/config.xml b/Tethering/res/values/config.xml
index effc24a..c999221 100644
--- a/Tethering/res/values/config.xml
+++ b/Tethering/res/values/config.xml
@@ -62,6 +62,13 @@
     <string-array translatable="false" name="config_tether_dhcp_range">
     </string-array>
 
+    <!-- Used to config periodic polls tether offload stats from tethering offload HAL to make the
+     data warnings work. 5000(ms) by default. If the device doesn't want to poll tether
+     offload stats, this should be -1. Note that this setting could be override by
+     runtime resource overlays.
+    -->
+    <integer translatable="false" name="config_tether_offload_poll_interval">5000</integer>
+
     <!-- Array of ConnectivityManager.TYPE_{BLUETOOTH, ETHERNET, MOBILE, MOBILE_DUN, MOBILE_HIPRI,
          WIFI} values allowable for tethering.
 
diff --git a/Tethering/res/values/overlayable.xml b/Tethering/res/values/overlayable.xml
index 16ae8ad..4c78a74 100644
--- a/Tethering/res/values/overlayable.xml
+++ b/Tethering/res/values/overlayable.xml
@@ -24,6 +24,7 @@
             <item type="array" name="config_tether_bluetooth_regexs"/>
             <item type="array" name="config_tether_dhcp_range"/>
             <item type="bool" name="config_tether_enable_legacy_dhcp_server"/>
+            <item type="integer" name="config_tether_offload_poll_interval"/>
             <item type="array" name="config_tether_upstream_types"/>
             <item type="bool" name="config_tether_upstream_automatic"/>
             <!-- Configuration values for tethering entitlement check -->
diff --git a/Tethering/src/com/android/networkstack/tethering/OffloadController.java b/Tethering/src/com/android/networkstack/tethering/OffloadController.java
index 1817f35..c063863 100644
--- a/Tethering/src/com/android/networkstack/tethering/OffloadController.java
+++ b/Tethering/src/com/android/networkstack/tethering/OffloadController.java
@@ -77,7 +77,8 @@
     private static final boolean DBG = false;
     private static final String ANYIP = "0.0.0.0";
     private static final ForwardedStats EMPTY_STATS = new ForwardedStats();
-    private static final int DEFAULT_PERFORM_POLL_INTERVAL_MS = 5000;
+    @VisibleForTesting
+    static final int DEFAULT_PERFORM_POLL_INTERVAL_MS = 5000;
 
     @VisibleForTesting
     enum StatsType {
diff --git a/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java b/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
index aeac437..9d4e747 100644
--- a/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
+++ b/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
@@ -78,6 +78,12 @@
     public static final String TETHER_ENABLE_LEGACY_DHCP_SERVER =
             "tether_enable_legacy_dhcp_server";
 
+    /**
+     * Default value that used to periodic polls tether offload stats from tethering offload HAL
+     * to make the data warnings work.
+     */
+    public static final int DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS = 5000;
+
     public final String[] tetherableUsbRegexs;
     public final String[] tetherableWifiRegexs;
     public final String[] tetherableWifiP2pRegexs;
@@ -96,6 +102,8 @@
 
     public final int activeDataSubId;
 
+    private final int mOffloadPollInterval;
+
     public TetheringConfiguration(Context ctx, SharedLog log, int id) {
         final SharedLog configLog = log.forSubComponent("config");
 
@@ -129,6 +137,10 @@
                 R.integer.config_mobile_hotspot_provision_check_period,
                 0 /* No periodic re-check */);
 
+        mOffloadPollInterval = getResourceInteger(res,
+                R.integer.config_tether_offload_poll_interval,
+                DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS);
+
         configLog.log(toString());
     }
 
@@ -189,6 +201,9 @@
         dumpStringArray(pw, "legacyDhcpRanges", legacyDhcpRanges);
         dumpStringArray(pw, "defaultIPv4DNS", defaultIPv4DNS);
 
+        pw.print("offloadPollInterval: ");
+        pw.println(mOffloadPollInterval);
+
         dumpStringArray(pw, "provisioningApp", provisioningApp);
         pw.print("provisioningAppNoUi: ");
         pw.println(provisioningAppNoUi);
@@ -208,6 +223,7 @@
                 makeString(tetherableBluetoothRegexs)));
         sj.add(String.format("isDunRequired:%s", isDunRequired));
         sj.add(String.format("chooseUpstreamAutomatically:%s", chooseUpstreamAutomatically));
+        sj.add(String.format("offloadPollInterval:%d", mOffloadPollInterval));
         sj.add(String.format("preferredUpstreamIfaceTypes:%s",
                 toIntArray(preferredUpstreamIfaceTypes)));
         sj.add(String.format("provisioningApp:%s", makeString(provisioningApp)));
@@ -246,6 +262,10 @@
         return (tm != null) ? tm.isTetheringApnRequired() : false;
     }
 
+    public int getOffloadPollInterval() {
+        return mOffloadPollInterval;
+    }
+
     private static Collection<Integer> getUpstreamIfaceTypes(Resources res, boolean dunRequired) {
         final int[] ifaceTypes = res.getIntArray(R.array.config_tether_upstream_types);
         final ArrayList<Integer> upstreamIfaceTypes = new ArrayList<>(ifaceTypes.length);
diff --git a/Tethering/tests/unit/src/com/android/networkstack/tethering/OffloadControllerTest.java b/Tethering/tests/unit/src/com/android/networkstack/tethering/OffloadControllerTest.java
index 088a663..f71cd21 100644
--- a/Tethering/tests/unit/src/com/android/networkstack/tethering/OffloadControllerTest.java
+++ b/Tethering/tests/unit/src/com/android/networkstack/tethering/OffloadControllerTest.java
@@ -26,18 +26,18 @@
 import static android.net.RouteInfo.RTN_UNICAST;
 import static android.provider.Settings.Global.TETHER_OFFLOAD_DISABLED;
 
+import static com.android.networkstack.tethering.OffloadController.DEFAULT_PERFORM_POLL_INTERVAL_MS;
 import static com.android.networkstack.tethering.OffloadController.StatsType.STATS_PER_IFACE;
 import static com.android.networkstack.tethering.OffloadController.StatsType.STATS_PER_UID;
 import static com.android.networkstack.tethering.OffloadHardwareInterface.ForwardedStats;
 import static com.android.testutils.MiscAssertsKt.assertContainsAll;
 import static com.android.testutils.MiscAssertsKt.assertThrows;
-import static com.android.testutils.NetworkStatsUtilsKt.orderInsensitiveEquals;
+import static com.android.testutils.NetworkStatsUtilsKt.assertNetworkStatsEquals;
 
 import static junit.framework.Assert.assertNotNull;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyObject;
@@ -63,10 +63,10 @@
 import android.net.NetworkStats;
 import android.net.NetworkStats.Entry;
 import android.net.RouteInfo;
-import android.net.netstats.provider.INetworkStatsProviderCallback;
+import android.net.netstats.provider.NetworkStatsProvider;
 import android.net.util.SharedLog;
 import android.os.Handler;
-import android.os.Looper;
+import android.os.test.TestLooper;
 import android.provider.Settings;
 import android.provider.Settings.SettingNotFoundException;
 import android.test.mock.MockContentResolver;
@@ -75,7 +75,7 @@
 import androidx.test.runner.AndroidJUnit4;
 
 import com.android.internal.util.test.FakeSettingsProvider;
-import com.android.testutils.HandlerUtilsKt;
+import com.android.testutils.TestableNetworkStatsProviderCbBinder;
 
 import org.junit.After;
 import org.junit.Before;
@@ -109,17 +109,19 @@
     @Mock private ApplicationInfo mApplicationInfo;
     @Mock private Context mContext;
     @Mock private NetworkStatsManager mStatsManager;
-    @Mock private INetworkStatsProviderCallback mTetherStatsProviderCb;
+    // Late init since methods must be called by the thread that created this object.
+    private TestableNetworkStatsProviderCbBinder mTetherStatsProviderCb;
     private OffloadController.OffloadTetheringStatsProvider mTetherStatsProvider;
     private final ArgumentCaptor<ArrayList> mStringArrayCaptor =
             ArgumentCaptor.forClass(ArrayList.class);
     private final ArgumentCaptor<OffloadHardwareInterface.ControlCallback> mControlCallbackCaptor =
             ArgumentCaptor.forClass(OffloadHardwareInterface.ControlCallback.class);
     private MockContentResolver mContentResolver;
+    private final TestLooper mTestLooper = new TestLooper();
     private OffloadController.Dependencies mDeps = new OffloadController.Dependencies() {
         @Override
         int getPerformPollInterval() {
-            return 0;
+            return -1; // Disabled.
         }
     };
 
@@ -150,12 +152,21 @@
         Settings.Global.putInt(mContentResolver, TETHER_OFFLOAD_DISABLED, 0);
     }
 
+    private void setOffloadPollInterval(int interval) {
+        mDeps = new OffloadController.Dependencies() {
+            @Override
+            int getPerformPollInterval() {
+                return interval;
+            }
+        };
+    }
+
     private void waitForIdle() {
-        HandlerUtilsKt.waitForIdle(new Handler(Looper.getMainLooper()), WAIT_FOR_IDLE_TIMEOUT);
+        mTestLooper.dispatchAll();
     }
 
     private OffloadController makeOffloadController() throws Exception {
-        OffloadController offload = new OffloadController(new Handler(Looper.getMainLooper()),
+        OffloadController offload = new OffloadController(new Handler(mTestLooper.getLooper()),
                 mHardware, mContentResolver, mStatsManager, new SharedLog("test"), mDeps);
         final ArgumentCaptor<OffloadController.OffloadTetheringStatsProvider>
                 tetherStatsProviderCaptor =
@@ -164,6 +175,7 @@
                 tetherStatsProviderCaptor.capture());
         mTetherStatsProvider = tetherStatsProviderCaptor.getValue();
         assertNotNull(mTetherStatsProvider);
+        mTetherStatsProviderCb = new TestableNetworkStatsProviderCbBinder();
         mTetherStatsProvider.setProviderCallbackBinder(mTetherStatsProviderCb);
         return offload;
     }
@@ -459,20 +471,12 @@
                 .addEntry(buildTestEntry(STATS_PER_UID, mobileIface, 999, 99999))
                 .addEntry(buildTestEntry(STATS_PER_UID, ethernetIface, 12345, 54321));
 
-        assertTrue(orderInsensitiveEquals(expectedIfaceStats, ifaceStats));
-        assertTrue(orderInsensitiveEquals(expectedUidStats, uidStats));
-
-        final ArgumentCaptor<NetworkStats> ifaceStatsCaptor = ArgumentCaptor.forClass(
-                NetworkStats.class);
-        final ArgumentCaptor<NetworkStats> uidStatsCaptor = ArgumentCaptor.forClass(
-                NetworkStats.class);
+        assertNetworkStatsEquals(expectedIfaceStats, ifaceStats);
+        assertNetworkStatsEquals(expectedUidStats, uidStats);
 
         // Force pushing stats update to verify the stats reported.
         mTetherStatsProvider.pushTetherStats();
-        verify(mTetherStatsProviderCb, times(1))
-                .notifyStatsUpdated(anyInt(), ifaceStatsCaptor.capture(), uidStatsCaptor.capture());
-        assertTrue(orderInsensitiveEquals(expectedIfaceStats, ifaceStatsCaptor.getValue()));
-        assertTrue(orderInsensitiveEquals(expectedUidStats, uidStatsCaptor.getValue()));
+        mTetherStatsProviderCb.expectNotifyStatsUpdated(expectedIfaceStats, expectedUidStats);
 
         when(mHardware.getForwardedStats(eq(ethernetIface))).thenReturn(
                 new ForwardedStats(100000, 100000));
@@ -498,11 +502,10 @@
                 .addEntry(buildTestEntry(STATS_PER_UID, mobileIface, 999, 99999))
                 .addEntry(buildTestEntry(STATS_PER_UID, ethernetIface, 112345, 154321));
 
-        assertTrue(orderInsensitiveEquals(expectedIfaceStatsAccu, ifaceStatsAccu));
-        assertTrue(orderInsensitiveEquals(expectedUidStatsAccu, uidStatsAccu));
+        assertNetworkStatsEquals(expectedIfaceStatsAccu, ifaceStatsAccu);
+        assertNetworkStatsEquals(expectedUidStatsAccu, uidStatsAccu);
 
         // Verify that only diff of stats is reported.
-        reset(mTetherStatsProviderCb);
         mTetherStatsProvider.pushTetherStats();
         final NetworkStats expectedIfaceStatsDiff = new NetworkStats(0L, 2)
                 .addEntry(buildTestEntry(STATS_PER_IFACE, mobileIface, 0, 0))
@@ -511,10 +514,8 @@
         final NetworkStats expectedUidStatsDiff = new NetworkStats(0L, 2)
                 .addEntry(buildTestEntry(STATS_PER_UID, mobileIface, 0, 0))
                 .addEntry(buildTestEntry(STATS_PER_UID, ethernetIface, 100000, 100000));
-        verify(mTetherStatsProviderCb, times(1))
-                .notifyStatsUpdated(anyInt(), ifaceStatsCaptor.capture(), uidStatsCaptor.capture());
-        assertTrue(orderInsensitiveEquals(expectedIfaceStatsDiff, ifaceStatsCaptor.getValue()));
-        assertTrue(orderInsensitiveEquals(expectedUidStatsDiff, uidStatsCaptor.getValue()));
+        mTetherStatsProviderCb.expectNotifyStatsUpdated(expectedIfaceStatsDiff,
+                expectedUidStatsDiff);
     }
 
     @Test
@@ -591,7 +592,7 @@
 
         OffloadHardwareInterface.ControlCallback callback = mControlCallbackCaptor.getValue();
         callback.onStoppedLimitReached();
-        verify(mTetherStatsProviderCb, times(1)).notifyStatsUpdated(anyInt(), any(), any());
+        mTetherStatsProviderCb.expectNotifyStatsUpdated();
     }
 
     @Test
@@ -695,8 +696,8 @@
         verify(mHardware, times(1)).getForwardedStats(eq(RMNET0));
         verify(mHardware, times(1)).getForwardedStats(eq(WLAN0));
         // TODO: verify the exact stats reported.
-        verify(mTetherStatsProviderCb, times(1)).notifyStatsUpdated(anyInt(), any(), any());
-        verifyNoMoreInteractions(mTetherStatsProviderCb);
+        mTetherStatsProviderCb.expectNotifyStatsUpdated();
+        mTetherStatsProviderCb.assertNoCallback();
         verifyNoMoreInteractions(mHardware);
     }
 
@@ -760,8 +761,8 @@
         // Verify forwarded stats behaviour.
         verify(mHardware, times(1)).getForwardedStats(eq(RMNET0));
         verify(mHardware, times(1)).getForwardedStats(eq(WLAN0));
-        verify(mTetherStatsProviderCb, times(1)).notifyStatsUpdated(anyInt(), any(), any());
-        verifyNoMoreInteractions(mTetherStatsProviderCb);
+        mTetherStatsProviderCb.expectNotifyStatsUpdated();
+        mTetherStatsProviderCb.assertNoCallback();
 
         // TODO: verify local prefixes and downstreams are also pushed to the HAL.
         verify(mHardware, times(1)).setLocalPrefixes(mStringArrayCaptor.capture());
@@ -780,4 +781,50 @@
         verifyNoMoreInteractions(mHardware);
     }
 
+    @Test
+    public void testOnSetAlert() throws Exception {
+        setupFunctioningHardwareInterface();
+        enableOffload();
+        setOffloadPollInterval(DEFAULT_PERFORM_POLL_INTERVAL_MS);
+        final OffloadController offload = makeOffloadController();
+        offload.start();
+
+        // Initialize with fake eth upstream.
+        final String ethernetIface = "eth1";
+        InOrder inOrder = inOrder(mHardware);
+        final LinkProperties lp = new LinkProperties();
+        lp.setInterfaceName(ethernetIface);
+        offload.setUpstreamLinkProperties(lp);
+        // Previous upstream was null, so no stats are fetched.
+        inOrder.verify(mHardware, never()).getForwardedStats(any());
+
+        // Verify that set quota to 0 will immediately triggers an callback.
+        mTetherStatsProvider.onSetAlert(0);
+        waitForIdle();
+        mTetherStatsProviderCb.expectNotifyAlertReached();
+
+        // Verify that notifyAlertReached never fired if quota is not yet reached.
+        when(mHardware.getForwardedStats(eq(ethernetIface))).thenReturn(
+                new ForwardedStats(0, 0));
+        mTetherStatsProvider.onSetAlert(100);
+        mTestLooper.moveTimeForward(DEFAULT_PERFORM_POLL_INTERVAL_MS);
+        waitForIdle();
+        mTetherStatsProviderCb.assertNoCallback();
+
+        // Verify that notifyAlertReached fired when quota is reached.
+        when(mHardware.getForwardedStats(eq(ethernetIface))).thenReturn(
+                new ForwardedStats(50, 50));
+        mTestLooper.moveTimeForward(DEFAULT_PERFORM_POLL_INTERVAL_MS);
+        waitForIdle();
+        mTetherStatsProviderCb.expectNotifyAlertReached();
+
+        // Verify that set quota with UNLIMITED won't trigger any callback, and won't fetch
+        // any stats since the polling is stopped.
+        reset(mHardware);
+        mTetherStatsProvider.onSetAlert(NetworkStatsProvider.QUOTA_UNLIMITED);
+        mTestLooper.moveTimeForward(DEFAULT_PERFORM_POLL_INTERVAL_MS);
+        waitForIdle();
+        mTetherStatsProviderCb.assertNoCallback();
+        verify(mHardware, never()).getForwardedStats(any());
+    }
 }
diff --git a/Tethering/tests/unit/src/com/android/networkstack/tethering/TetheringConfigurationTest.java b/Tethering/tests/unit/src/com/android/networkstack/tethering/TetheringConfigurationTest.java
index 07ddea4..e8ba5b8 100644
--- a/Tethering/tests/unit/src/com/android/networkstack/tethering/TetheringConfigurationTest.java
+++ b/Tethering/tests/unit/src/com/android/networkstack/tethering/TetheringConfigurationTest.java
@@ -115,6 +115,8 @@
 
         when(mResources.getStringArray(R.array.config_tether_dhcp_range)).thenReturn(
                 new String[0]);
+        when(mResources.getInteger(R.integer.config_tether_offload_poll_interval)).thenReturn(
+                TetheringConfiguration.DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS);
         when(mResources.getStringArray(R.array.config_tether_usb_regexs)).thenReturn(new String[0]);
         when(mResources.getStringArray(R.array.config_tether_wifi_regexs))
                 .thenReturn(new String[]{ "test_wlan\\d" });
@@ -314,6 +316,23 @@
     }
 
     @Test
+    public void testOffloadIntervalByResource() {
+        final TetheringConfiguration intervalByDefault =
+                new TetheringConfiguration(mMockContext, mLog, INVALID_SUBSCRIPTION_ID);
+        assertEquals(TetheringConfiguration.DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS,
+                intervalByDefault.getOffloadPollInterval());
+
+        final int[] testOverrides = {0, 3000, -1};
+        for (final int override : testOverrides) {
+            when(mResources.getInteger(R.integer.config_tether_offload_poll_interval)).thenReturn(
+                    override);
+            final TetheringConfiguration overrideByRes =
+                    new TetheringConfiguration(mMockContext, mLog, INVALID_SUBSCRIPTION_ID);
+            assertEquals(override, overrideByRes.getOffloadPollInterval());
+        }
+    }
+
+    @Test
     public void testGetResourcesBySubId() {
         setUpResourceForSubId();
         final TetheringConfiguration cfg = new TetheringConfiguration(