diff --git a/include/seastar/core/semaphore.hh b/include/seastar/core/semaphore.hh
index ad3a0a18014..bf125413227 100644
--- a/include/seastar/core/semaphore.hh
+++ b/include/seastar/core/semaphore.hh
@@ -482,8 +482,11 @@ public:
     semaphore_units(semaphore_units&& o) noexcept : _sem(o._sem), _n(std::exchange(o._n, 0)) {
     }
     semaphore_units& operator=(semaphore_units&& o) noexcept {
-        _sem = o._sem;
-        _n = std::exchange(o._n, 0);
+        if (this != &o) {
+            return_all();
+            _sem = o._sem;
+            _n = std::exchange(o._n, 0);
+        }
         return *this;
     }
     semaphore_units(const semaphore_units&) = delete;
diff --git a/tests/unit/semaphore_test.cc b/tests/unit/semaphore_test.cc
index b62b6798795..b9a15db7f60 100644
--- a/tests/unit/semaphore_test.cc
+++ b/tests/unit/semaphore_test.cc
@@ -262,7 +262,7 @@ SEASTAR_TEST_CASE(test_with_semaphore) {
     });
 }
 
-SEASTAR_THREAD_TEST_CASE(test_semaphore_units_splitting) {
+SEASTAR_THREAD_TEST_CASE(test_semaphore_units_valid_splitting) {
     auto sm = semaphore(2);
     auto units = get_units(sm, 2, 1min).get0();
     {
@@ -272,25 +272,32 @@ SEASTAR_THREAD_TEST_CASE(test_semaphore_units_splitting) {
         BOOST_REQUIRE_EQUAL(sm.available_units(), 0);
     }
     BOOST_REQUIRE_EQUAL(sm.available_units(), 1);
-    units.~semaphore_units();
-    units = get_units(sm, 2, 1min).get0();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_semaphore_units_invalid_splitting) {
+    auto sm = semaphore(2);
+    auto units = get_units(sm, 2, 1min).get0();
     BOOST_REQUIRE_EQUAL(sm.available_units(), 0);
     BOOST_REQUIRE_THROW(units.split(10), std::invalid_argument);
     BOOST_REQUIRE_EQUAL(sm.available_units(), 0);
 }
 
-SEASTAR_THREAD_TEST_CASE(test_semaphore_units_return) {
+SEASTAR_THREAD_TEST_CASE(test_semaphore_units_return_when_destroyed) {
     auto sm = semaphore(3);
+  {
     auto units = get_units(sm, 3, 1min).get0();
     BOOST_REQUIRE_EQUAL(units.count(), 3);
     BOOST_REQUIRE_EQUAL(sm.available_units(), 0);
     BOOST_REQUIRE_EQUAL(units.return_units(1), 2);
     BOOST_REQUIRE_EQUAL(units.count(), 2);
     BOOST_REQUIRE_EQUAL(sm.available_units(), 1);
-    units.~semaphore_units();
+  }
     BOOST_REQUIRE_EQUAL(sm.available_units(), 3);
+}
 
-    units = get_units(sm, 2, 1min).get0();
+SEASTAR_THREAD_TEST_CASE(test_semaphore_units_return_all) {
+    auto sm = semaphore(3);
+    auto units = get_units(sm, 2, 1min).get0();
     BOOST_REQUIRE_EQUAL(sm.available_units(), 1);
     BOOST_REQUIRE_THROW(units.return_units(10), std::invalid_argument);
     BOOST_REQUIRE_EQUAL(sm.available_units(), 1);
@@ -431,3 +438,17 @@ SEASTAR_THREAD_TEST_CASE(test_semaphore_abort_before_wait) {
     BOOST_CHECK_THROW(fut1.get(), semaphore_aborted);
     BOOST_REQUIRE_EQUAL(x, 0);
 }
+
+SEASTAR_THREAD_TEST_CASE(test_reassigned_units_are_returned) {
+    auto sem0 = semaphore(1);
+    auto sem1 = semaphore(1);
+    auto units = get_units(sem0, 1).get();
+    auto wait = sem0.wait(1);
+    BOOST_REQUIRE(!wait.available());
+    units = get_units(sem1, 1).get();
+    timer t([] { abort(); });
+    t.arm(1s);
+    // will hang if units are not returned when reassigned
+    wait.get();
+    t.cancel();
+}