diff --git a/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadManager.java b/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadManager.java
index c3cf0bdc24..66b2a7cf91 100644
--- a/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadManager.java
+++ b/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadManager.java
@@ -75,6 +75,16 @@ public final class DownloadManager {
      */
     default void onInitialized(DownloadManager downloadManager) {}
 
+    /**
+     * Called when downloads are ({@link #pauseDownloads() paused} or {@link #resumeDownloads()
+     * resumed}.
+     *
+     * @param downloadManager The reporting instance.
+     * @param downloadsPaused Whether downloads are currently paused.
+     */
+    default void onDownloadsPausedChanged(
+        DownloadManager downloadManager, boolean downloadsPaused) {}
+
     /**
      * Called when the state of a download changes.
      *
@@ -110,6 +120,19 @@ public final class DownloadManager {
         DownloadManager downloadManager,
         Requirements requirements,
         @Requirements.RequirementFlags int notMetRequirements) {}
+
+    /**
+     * Called when there is a change in whether this manager has one or more downloads that are not
+     * progressing for the sole reason that the {@link #getRequirements() Requirements} are not met.
+     * See {@link #isWaitingForRequirements()} for more information.
+     *
+     * @param downloadManager The reporting instance.
+     * @param waitingForRequirements Whether this manager has one or more downloads that are not
+     *     progressing for the sole reason that the {@link #getRequirements() Requirements} are not
+     *     met.
+     */
+    default void onWaitingForRequirementsChanged(
+        DownloadManager downloadManager, boolean waitingForRequirements) {}
   }
 
   /** The default maximum number of parallel downloads. */
@@ -155,6 +178,7 @@ public final class DownloadManager {
   private int maxParallelDownloads;
   private int minRetryCount;
   private int notMetRequirements;
+  private boolean waitingForRequirements;
   private List<Download> downloads;
   private RequirementsWatcher requirementsWatcher;
 
@@ -238,17 +262,16 @@ public final class DownloadManager {
 
   /**
    * Returns whether this manager has one or more downloads that are not progressing for the sole
-   * reason that the {@link #getRequirements() Requirements} are not met.
+   * reason that the {@link #getRequirements() Requirements} are not met. This is true if:
+   *
+   * <ul>
+   *   <li>The {@link #getRequirements() Requirements} are not met.
+   *   <li>The downloads are not paused (i.e. {@link #getDownloadsPaused()} is {@code false}).
+   *   <li>There are downloads in the {@link Download#STATE_QUEUED queued state}.
+   * </ul>
    */
   public boolean isWaitingForRequirements() {
-    if (!downloadsPaused && notMetRequirements != 0) {
-      for (int i = 0; i < downloads.size(); i++) {
-        if (downloads.get(i).state == STATE_QUEUED) {
-          return true;
-        }
-      }
-    }
-    return false;
+    return waitingForRequirements;
   }
 
   /**
@@ -281,7 +304,7 @@ public final class DownloadManager {
    */
   @Requirements.RequirementFlags
   public int getNotMetRequirements() {
-    return getRequirements().getNotMetRequirements(context);
+    return notMetRequirements;
   }
 
   /**
@@ -374,29 +397,15 @@ public final class DownloadManager {
    * {@link Download#stopReason stopReasons}.
    */
   public void resumeDownloads() {
-    if (!downloadsPaused) {
-      return;
-    }
-    downloadsPaused = false;
-    pendingMessages++;
-    internalHandler
-        .obtainMessage(MSG_SET_DOWNLOADS_PAUSED, /* downloadsPaused */ 0, /* unused */ 0)
-        .sendToTarget();
+    setDownloadsPaused(/* downloadsPaused= */ false);
   }
 
   /**
-   * Pauses downloads. Downloads that would otherwise be making progress transition to {@link
+   * Pauses downloads. Downloads that would otherwise be making progress will transition to {@link
    * Download#STATE_QUEUED}.
    */
   public void pauseDownloads() {
-    if (downloadsPaused) {
-      return;
-    }
-    downloadsPaused = true;
-    pendingMessages++;
-    internalHandler
-        .obtainMessage(MSG_SET_DOWNLOADS_PAUSED, /* downloadsPaused */ 1, /* unused */ 0)
-        .sendToTarget();
+    setDownloadsPaused(/* downloadsPaused= */ true);
   }
 
   /**
@@ -481,6 +490,26 @@ public final class DownloadManager {
       pendingMessages = 0;
       activeTaskCount = 0;
       initialized = false;
+      notMetRequirements = 0;
+      waitingForRequirements = false;
+    }
+  }
+
+  private void setDownloadsPaused(boolean downloadsPaused) {
+    if (this.downloadsPaused == downloadsPaused) {
+      return;
+    }
+    this.downloadsPaused = downloadsPaused;
+    pendingMessages++;
+    internalHandler
+        .obtainMessage(MSG_SET_DOWNLOADS_PAUSED, downloadsPaused ? 1 : 0, /* unused */ 0)
+        .sendToTarget();
+    boolean waitingForRequirementsChanged = updateWaitingForRequirements();
+    for (Listener listener : listeners) {
+      listener.onDownloadsPausedChanged(this, downloadsPaused);
+    }
+    if (waitingForRequirementsChanged) {
+      notifyWaitingForRequirementsChanged();
     }
   }
 
@@ -488,17 +517,41 @@ public final class DownloadManager {
       RequirementsWatcher requirementsWatcher,
       @Requirements.RequirementFlags int notMetRequirements) {
     Requirements requirements = requirementsWatcher.getRequirements();
+    if (this.notMetRequirements != notMetRequirements) {
+      this.notMetRequirements = notMetRequirements;
+      pendingMessages++;
+      internalHandler
+          .obtainMessage(MSG_SET_NOT_MET_REQUIREMENTS, notMetRequirements, /* unused */ 0)
+          .sendToTarget();
+    }
+    boolean waitingForRequirementsChanged = updateWaitingForRequirements();
     for (Listener listener : listeners) {
       listener.onRequirementsStateChanged(this, requirements, notMetRequirements);
     }
-    if (this.notMetRequirements == notMetRequirements) {
-      return;
+    if (waitingForRequirementsChanged) {
+      notifyWaitingForRequirementsChanged();
+    }
+  }
+
+  private boolean updateWaitingForRequirements() {
+    boolean waitingForRequirements = false;
+    if (!downloadsPaused && notMetRequirements != 0) {
+      for (int i = 0; i < downloads.size(); i++) {
+        if (downloads.get(i).state == STATE_QUEUED) {
+          waitingForRequirements = true;
+          break;
+        }
+      }
+    }
+    boolean waitingForRequirementsChanged = this.waitingForRequirements != waitingForRequirements;
+    this.waitingForRequirements = waitingForRequirements;
+    return waitingForRequirementsChanged;
+  }
+
+  private void notifyWaitingForRequirementsChanged() {
+    for (Listener listener : listeners) {
+      listener.onWaitingForRequirementsChanged(this, waitingForRequirements);
     }
-    this.notMetRequirements = notMetRequirements;
-    pendingMessages++;
-    internalHandler
-        .obtainMessage(MSG_SET_NOT_MET_REQUIREMENTS, notMetRequirements, /* unused */ 0)
-        .sendToTarget();
   }
 
   // Main thread message handling.
@@ -528,14 +581,19 @@ public final class DownloadManager {
   private void onInitialized(List<Download> downloads) {
     initialized = true;
     this.downloads = Collections.unmodifiableList(downloads);
+    boolean waitingForRequirementsChanged = updateWaitingForRequirements();
     for (Listener listener : listeners) {
       listener.onInitialized(DownloadManager.this);
     }
+    if (waitingForRequirementsChanged) {
+      notifyWaitingForRequirementsChanged();
+    }
   }
 
   private void onDownloadUpdate(DownloadUpdate update) {
     downloads = Collections.unmodifiableList(update.downloads);
     Download updatedDownload = update.download;
+    boolean waitingForRequirementsChanged = updateWaitingForRequirements();
     if (update.isRemove) {
       for (Listener listener : listeners) {
         listener.onDownloadRemoved(this, updatedDownload);
@@ -545,6 +603,9 @@ public final class DownloadManager {
         listener.onDownloadChanged(this, updatedDownload);
       }
     }
+    if (waitingForRequirementsChanged) {
+      notifyWaitingForRequirementsChanged();
+    }
   }
 
   private void onMessageProcessed(int processedMessageCount, int activeTaskCount) {
diff --git a/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadService.java b/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadService.java
index 6377a9e6ff..b1ab5ac7c6 100644
--- a/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadService.java
+++ b/library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadService.java
@@ -35,7 +35,6 @@ import com.google.android.exoplayer2.util.Util;
 import java.util.HashMap;
 import java.util.List;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
-import org.checkerframework.checker.nullness.qual.RequiresNonNull;
 
 /** A {@link Service} for downloading media. */
 public abstract class DownloadService extends Service {
@@ -183,6 +182,7 @@ public abstract class DownloadService extends Service {
   private int lastStartId;
   private boolean startedInForeground;
   private boolean taskRemoved;
+  private boolean isStopped;
   private boolean isDestroyed;
 
   /**
@@ -666,6 +666,8 @@ public abstract class DownloadService extends Service {
       // From API level 26, services started in the foreground are required to show a notification.
       foregroundNotificationUpdater.showNotificationIfNotAlready();
     }
+
+    isStopped = false;
     if (downloadManager.isIdle()) {
       stop();
     }
@@ -768,7 +770,7 @@ public abstract class DownloadService extends Service {
   private void notifyDownloads(List<Download> downloads) {
     if (foregroundNotificationUpdater != null) {
       for (int i = 0; i < downloads.size(); i++) {
-        if (needsForegroundNotification(downloads.get(i).state)) {
+        if (needsStartedService(downloads.get(i).state)) {
           foregroundNotificationUpdater.startPeriodicUpdates();
           break;
         }
@@ -785,7 +787,7 @@ public abstract class DownloadService extends Service {
   private void notifyDownloadChanged(Download download) {
     onDownloadChanged(download);
     if (foregroundNotificationUpdater != null) {
-      if (needsForegroundNotification(download.state)) {
+      if (needsStartedService(download.state)) {
         foregroundNotificationUpdater.startPeriodicUpdates();
       } else {
         foregroundNotificationUpdater.invalidate();
@@ -806,18 +808,24 @@ public abstract class DownloadService extends Service {
     }
   }
 
+  /** Returns whether the service is stopped. */
+  private boolean isStopped() {
+    return isStopped;
+  }
+
   private void stop() {
     if (foregroundNotificationUpdater != null) {
       foregroundNotificationUpdater.stopPeriodicUpdates();
     }
     if (Util.SDK_INT < 28 && taskRemoved) { // See [Internal: b/74248644].
       stopSelf();
+      isStopped = true;
     } else {
-      stopSelfResult(lastStartId);
+      isStopped |= stopSelfResult(lastStartId);
     }
   }
 
-  private static boolean needsForegroundNotification(@Download.State int state) {
+  private static boolean needsStartedService(@Download.State int state) {
     return state == Download.STATE_DOWNLOADING
         || state == Download.STATE_REMOVING
         || state == Download.STATE_RESTARTING;
@@ -910,10 +918,7 @@ public abstract class DownloadService extends Service {
       this.scheduler = scheduler;
       this.serviceClass = serviceClass;
       downloadManager.addListener(this);
-      if (this.scheduler != null) {
-        Requirements requirements = downloadManager.getRequirements();
-        setSchedulerEnabled(/* enabled= */ !requirements.checkRequirements(context), requirements);
-      }
+      updateScheduler();
     }
 
     public void attachService(DownloadService downloadService) {
@@ -953,6 +958,14 @@ public abstract class DownloadService extends Service {
       if (downloadService != null) {
         downloadService.notifyDownloadChanged(download);
       }
+      if (serviceMayNeedRestart() && needsStartedService(download.state)) {
+        // This shouldn't happen unless (a) application code is changing the downloads by calling
+        // the DownloadManager directly rather than sending actions through the service, or (b) if
+        // the service is background only and a previous attempt to start it was prevented. Try and
+        // restart the service to robust against such cases.
+        Log.w(TAG, "DownloadService wasn't running. Restarting.");
+        restartService();
+      }
     }
 
     @Override
@@ -970,24 +983,31 @@ public abstract class DownloadService extends Service {
     }
 
     @Override
-    public void onRequirementsStateChanged(
-        DownloadManager downloadManager,
-        Requirements requirements,
-        @Requirements.RequirementFlags int notMetRequirements) {
-      boolean requirementsMet = notMetRequirements == 0;
-      // TODO: Fix this logic to only start the service if the DownloadManager is actually going to
-      // make progress (in addition to the requirements being met, it also needs to be not paused
-      // and have some current downloads).
-      if (downloadService == null && requirementsMet) {
-        restartService();
-      }
-      if (scheduler != null) {
-        setSchedulerEnabled(/* enabled= */ !requirementsMet, requirements);
+    public void onWaitingForRequirementsChanged(
+        DownloadManager downloadManager, boolean waitingForRequirements) {
+      if (!waitingForRequirements
+          && !downloadManager.getDownloadsPaused()
+          && serviceMayNeedRestart()) {
+        // We're no longer waiting for requirements and downloads aren't paused, meaning the manager
+        // will be able to resume downloads that are currently queued. If there exist queued
+        // downloads then we should ensure the service is started.
+        List<Download> downloads = downloadManager.getCurrentDownloads();
+        for (int i = 0; i < downloads.size(); i++) {
+          if (downloads.get(i).state == Download.STATE_QUEUED) {
+            restartService();
+            break;
+          }
+        }
       }
+      updateScheduler();
     }
 
     // Internal methods.
 
+    private boolean serviceMayNeedRestart() {
+      return downloadService == null || downloadService.isStopped();
+    }
+
     private void restartService() {
       if (foregroundAllowed) {
         Intent intent = getIntent(context, serviceClass, DownloadService.ACTION_RESTART);
@@ -1001,24 +1021,24 @@ public abstract class DownloadService extends Service {
         } catch (IllegalArgumentException e) {
           // The process is classed as idle by the platform. Starting a background service is not
           // allowed in this state.
+          Log.w(TAG, "Failed to restart DownloadService (process is idle).");
         }
       }
     }
 
-    // TODO: Fix callers to this method so that the scheduler is only enabled if the DownloadManager
-    // would actually make progress were the requirements met (or if it's not initialized yet, in
-    // which case we should be cautious until we know better). To implement this properly it'll be
-    // necessary to call this method from some additional places.
-    @RequiresNonNull("scheduler")
-    private void setSchedulerEnabled(boolean enabled, Requirements requirements) {
-      if (!enabled) {
-        scheduler.cancel();
-      } else {
+    private void updateScheduler() {
+      if (scheduler == null) {
+        return;
+      }
+      if (downloadManager.isWaitingForRequirements()) {
         String servicePackage = context.getPackageName();
+        Requirements requirements = downloadManager.getRequirements();
         boolean success = scheduler.schedule(requirements, servicePackage, ACTION_RESTART);
         if (!success) {
           Log.e(TAG, "Scheduling downloads failed.");
         }
+      } else {
+        scheduler.cancel();
       }
     }
   }