diff --git a/log4j-cassandra/src/main/java/org/apache/logging/log4j/cassandra/CassandraManager.java b/log4j-cassandra/src/main/java/org/apache/logging/log4j/cassandra/CassandraManager.java
index 64c5221044e..2410c1944fa 100644
--- a/log4j-cassandra/src/main/java/org/apache/logging/log4j/cassandra/CassandraManager.java
+++ b/log4j-cassandra/src/main/java/org/apache/logging/log4j/cassandra/CassandraManager.java
@@ -81,8 +81,14 @@ protected void startupInternal() throws Exception {
 
     @Override
     protected boolean shutdownInternal() throws Exception {
-        session.close();
-        cluster.close();
+        // session may be null if startupInternal failed (or was never called); cluster is created in the factory
+        try {
+            if (session != null) {
+                session.close();
+            }
+        } finally {
+            cluster.close();
+        }
         return true;
     }
 
diff --git a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManagerTest.java b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManagerTest.java
index 38fad66231c..1181f099da1 100644
--- a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManagerTest.java
+++ b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManagerTest.java
@@ -23,6 +23,7 @@
 import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.BDDMockito.then;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
@@ -57,7 +58,7 @@ protected boolean shutdownInternal() {
         }
 
         @Override
-        protected void startupInternal() {
+        protected void startupInternal() throws Exception {
             // noop
         }
 
@@ -67,6 +68,19 @@ protected void writeInternal(final LogEvent event, final Serializable serializab
         }
     }
 
+    /** Stub whose {@link #startupInternal()} fails after (simulated) partial resource acquisition. */
+    private static class FailingStartupDatabaseManager extends StubDatabaseManager {
+
+        private FailingStartupDatabaseManager(final String name, final int bufferSize) {
+            super(name, bufferSize);
+        }
+
+        @Override
+        protected void startupInternal() throws Exception {
+            throw new Exception("simulated startup failure");
+        }
+    }
+
     private AbstractDatabaseManager manager;
 
     public void setUp(final String name, final int buffer) {
@@ -272,4 +286,59 @@ void testToString02() {
 
         assertEquals("bufferSize=12, anotherKey02=coolValue02", manager.toString(), "The string is not correct.");
     }
+
+    /**
+     * After startupInternal fails, the manager must not accept writes (avoids per-event NPEs on unassigned state).
+     */
+    @Test
+    void testFailedStartupSkipsWrite() throws Exception {
+        manager = spy(new FailingStartupDatabaseManager("failedStartupWrite", 0));
+
+        manager.startup();
+        assertFalse(manager.isRunning(), "Manager must not be running after startupInternal fails.");
+        then(manager).should().startupInternal();
+
+        final LogEvent event1 = mock(LogEvent.class);
+        final LogEvent event2 = mock(LogEvent.class);
+        manager.write(event1, null);
+        manager.write(event2, null);
+
+        then(manager).should(never()).writeThrough(same(event1), isNull());
+        then(manager).should(never()).writeThrough(same(event2), isNull());
+        then(manager).should(never()).writeInternal(same(event1), isNull());
+        then(manager).should(never()).writeInternal(same(event2), isNull());
+        then(manager).should(never()).connectAndStart();
+    }
+
+    /**
+     * After startupInternal fails, shutdown must still invoke shutdownInternal so partial resources are released.
+     */
+    @Test
+    void testFailedStartupStillShutsDown() throws Exception {
+        manager = spy(new FailingStartupDatabaseManager("failedStartupShutdown", 0));
+
+        manager.startup();
+        assertFalse(manager.isRunning(), "Manager must not be running after startupInternal fails.");
+
+        assertTrue(manager.shutdown(), "shutdown should complete after a failed startup.");
+        then(manager).should().shutdownInternal();
+        assertFalse(manager.isRunning(), "Manager must remain not running after shutdown.");
+
+        // second shutdown must not call shutdownInternal again
+        reset(manager);
+        assertTrue(manager.shutdown());
+        then(manager).should(never()).shutdownInternal();
+    }
+
+    /**
+     * Shutdown without a prior successful startup still runs shutdownInternal once (factory-time resources).
+     */
+    @Test
+    void testShutdownWithoutStartupStillRunsShutdownInternal() throws Exception {
+        setUp("neverStarted", 0);
+
+        assertFalse(manager.isRunning());
+        assertTrue(manager.shutdown());
+        then(manager).should().shutdownInternal();
+    }
 }
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManager.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManager.java
index f166ec2ce88..23ff92b68e6 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManager.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/db/AbstractDatabaseManager.java
@@ -106,6 +106,19 @@ protected static <M extends AbstractDatabaseManager, T extends AbstractFactoryDa
 
     private boolean running;
 
+    /**
+     * Whether {@link #shutdownInternal()} has already been invoked for this manager instance.
+     * Tracks completed shutdown independently of {@link #running} so resources acquired during a
+     * failed {@link #startupInternal()} can still be released.
+     */
+    private boolean shutDown;
+
+    /**
+     * Whether we already logged that writes are being skipped while the manager is not running.
+     * Avoids flooding the status logger with one message per event after a failed startup.
+     */
+    private boolean writeWhileNotRunningLogged;
+
     /**
      * Constructs the base manager.
      *
@@ -222,12 +235,17 @@ public final boolean releaseSub(final long timeout, final TimeUnit timeUnit) {
      * This method is called from the {@link #close()} method when the appender is stopped or the appender's manager
      * is replaced. If it has not already been called, it calls {@link #shutdownInternal()} and catches any exceptions
      * it might throw.
+     * <p>
+     * {@link #shutdownInternal()} is invoked even when {@link #isRunning()} is {@code false}, so implementations can
+     * release resources acquired during a failed {@link #startupInternal()} (or before startup).
+     * </p>
      * @return true if all resources were closed normally, false otherwise.
      */
     public final synchronized boolean shutdown() {
         boolean closed = true;
         this.flush();
-        if (this.isRunning()) {
+        if (!this.shutDown) {
+            this.shutDown = true;
             try {
                 closed &= this.shutdownInternal();
             } catch (final Exception e) {
@@ -242,8 +260,9 @@ public final synchronized boolean shutdown() {
 
     /**
      * Implementations should implement this method to perform any proprietary disconnection / shutdown operations. This
-     * method will never be called twice on the same instance, and it will only be called <em>after</em>
-     * {@link #startupInternal()}. It is safe to throw any exceptions from this method. This method does not
+     * method will never be called twice on the same instance between successful startups. It may be called after a
+     * failed {@link #startupInternal()} (or even if startup was never attempted), so implementations must tolerate
+     * partially initialized state. It is safe to throw any exceptions from this method. This method does not
      * necessarily disconnect from the database for the same reasons outlined in {@link #startupInternal()}.
      * @return true if all resources were closed normally, false otherwise.
      */
@@ -258,6 +277,8 @@ public final synchronized void startup() {
             try {
                 this.startupInternal();
                 this.running = true;
+                this.shutDown = false;
+                this.writeWhileNotRunningLogged = false;
             } catch (final Exception e) {
                 logError("Could not perform database startup operations", e);
             }
@@ -290,11 +311,25 @@ public final synchronized void write(final LogEvent event) {
 
     /**
      * This method manages buffering and writing of events.
+     * <p>
+     * If the manager is not running (for example because {@link #startupInternal()} failed), the event is dropped and
+     * a single status warning is logged rather than attempting a write that would likely fail with an NPE per event.
+     * </p>
      *
      * @param event The event to write to the database.
      * @param serializable Serializable event
      */
     public final synchronized void write(final LogEvent event, final Serializable serializable) {
+        if (!this.isRunning()) {
+            if (!this.writeWhileNotRunningLogged) {
+                this.writeWhileNotRunningLogged = true;
+                LOGGER.warn(
+                        "{} {} is not running; skipping database write until startup succeeds",
+                        getClass().getSimpleName(),
+                        getName());
+            }
+            return;
+        }
         if (isBuffered()) {
             buffer(event);
         } else {
diff --git a/src/changelog/.2.x.x/4241_fix_AbstractDatabaseManager_failed_startup.xml b/src/changelog/.2.x.x/4241_fix_AbstractDatabaseManager_failed_startup.xml
new file mode 100644
index 00000000000..980f6a76ab8
--- /dev/null
+++ b/src/changelog/.2.x.x/4241_fix_AbstractDatabaseManager_failed_startup.xml
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<entry xmlns="https://logging.apache.org/xml/ns"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="
+           https://logging.apache.org/xml/ns
+           https://logging.apache.org/xml/ns/log4j-changelog-0.xsd"
+       type="fixed">
+  <issue id="4241" link="https://github.com/apache/logging-log4j2/issues/4241"/>
+  <description format="asciidoc">
+    Fix `AbstractDatabaseManager` so a manager whose startup failed no longer accepts writes (which could NPE per event) and still runs `shutdownInternal()` to release resources acquired during startup.
+  </description>
+</entry>