diff --git a/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/Logging.java b/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/Logging.java
index e8b39e3e06ec..c262fdedae26 100644
--- a/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/Logging.java
+++ b/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/Logging.java
@@ -186,8 +186,15 @@ public static EntryListOption filter(String filter) {
   Synchronicity getWriteSynchronicity();
 
   /**
-   * Sets flush severity for asynchronous logging writes. Default is ERROR. Logs will be immediately
-   * written out for entries at or higher than flush severity.
+   * Sets flush severity for asynchronous logging writes. It is disabled by default, enabled when
+   * this method is called with not null value. Logs will be immediately written out for entries at
+   * or higher than flush severity.
+   *
+   * <p>Enabling this can cause the leaking and hanging threads, see BUG(2796) BUG(3880). However
+   * you can explicitly call {@link #flush}.
+   *
+   * <p>TODO: Enable this by default once functionality to trigger rpc is available in generated
+   * code.
    */
   void setFlushSeverity(Severity flushSeverity);
 
@@ -664,8 +671,8 @@ ApiFuture<AsyncPage<MonitoredResourceDescriptor>> listMonitoredResourceDescripto
   /**
    * Flushes any pending asynchronous logging writes. Logs are automatically flushed based on time
    * and message count that be configured via {@link com.google.api.gax.batching.BatchingSettings},
-   * Logs are also flushed if at or above flush severity, see {@link #setFlushSeverity}. Logging
-   * frameworks require support for an explicit flush. See usage in the java.util.logging
+   * Logs are also flushed if enabled, at or above flush severity, see {@link #setFlushSeverity}.
+   * Logging frameworks require support for an explicit flush. See usage in the java.util.logging
    * handler{@link LoggingHandler}.
    */
   void flush();
diff --git a/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/LoggingImpl.java b/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/LoggingImpl.java
index 07bb2cd70e69..7c572af069b5 100644
--- a/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/LoggingImpl.java
+++ b/google-cloud-clients/google-cloud-logging/src/main/java/com/google/cloud/logging/LoggingImpl.java
@@ -75,16 +75,19 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 class LoggingImpl extends BaseService<LoggingOptions> implements Logging {
 
+  private static final int FLUSH_WAIT_TIMEOUT_SECONDS = 6;
   private final LoggingRpc rpc;
   private final Object writeLock = new Object();
   private final Set<ApiFuture<Void>> pendingWrites =
       Collections.newSetFromMap(new IdentityHashMap<ApiFuture<Void>, Boolean>());
 
   private volatile Synchronicity writeSynchronicity = Synchronicity.ASYNC;
-  private volatile Severity flushSeverity = Severity.ERROR;
+  private volatile Severity flushSeverity = null;
   private boolean closed;
 
   private static final Function<Empty, Boolean> EMPTY_TO_BOOLEAN_FUNCTION =
@@ -553,11 +556,13 @@ public void write(Iterable<LogEntry> logEntries, WriteOption... options) {
 
     try {
       writeLogEntries(logEntries, options);
-      for (LogEntry logEntry : logEntries) {
-        // flush pending writes if log severity at or above flush severity
-        if (logEntry.getSeverity().compareTo(flushSeverity) >= 0) {
-          flush();
-          break;
+      if (flushSeverity != null) {
+        for (LogEntry logEntry : logEntries) {
+          // flush pending writes if log severity at or above flush severity
+          if (logEntry.getSeverity().compareTo(flushSeverity) >= 0) {
+            flush();
+            break;
+          }
         }
       }
     } finally {
@@ -574,8 +579,8 @@ public void flush() {
     }
 
     try {
-      ApiFutures.allAsList(writesToFlush).get();
-    } catch (InterruptedException | ExecutionException e) {
+      ApiFutures.allAsList(writesToFlush).get(FLUSH_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+    } catch (InterruptedException | ExecutionException | TimeoutException e) {
       throw new RuntimeException(e);
     }
   }
diff --git a/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/BaseSystemTest.java b/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/BaseSystemTest.java
index 4360d6724ab5..d3f53cdd3bdf 100644
--- a/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/BaseSystemTest.java
+++ b/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/BaseSystemTest.java
@@ -229,6 +229,7 @@ public void testWriteAndListLogEntries() throws InterruptedException {
             .build();
     logging().write(ImmutableList.of(firstEntry));
     logging().write(ImmutableList.of(secondEntry));
+    logging().flush();
     String filter = createEqualityFilter("logName", logName);
     EntryListOption[] options = {EntryListOption.filter(filter), EntryListOption.pageSize(1)};
     Page<LogEntry> page = logging().listLogEntries(options);
diff --git a/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/LoggingImplTest.java b/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/LoggingImplTest.java
index fd1335486226..7c711cd44a78 100644
--- a/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/LoggingImplTest.java
+++ b/google-cloud-clients/google-cloud-logging/src/test/java/com/google/cloud/logging/LoggingImplTest.java
@@ -1188,6 +1188,41 @@ public void testWriteLogEntries() {
     logging.write(ImmutableList.of(LOG_ENTRY1, LOG_ENTRY2));
   }
 
+  @Test
+  public void testWriteLogEntriesDoesnotEnableFlushByDefault() {
+    WriteLogEntriesRequest request =
+        WriteLogEntriesRequest.newBuilder()
+            .addAllEntries(
+                Iterables.transform(
+                    ImmutableList.of(
+                        LOG_ENTRY1, LOG_ENTRY2.toBuilder().setSeverity(Severity.EMERGENCY).build()),
+                    LogEntry.toPbFunction(PROJECT)))
+            .build();
+    ApiFuture<WriteLogEntriesResponse> apiFuture = SettableApiFuture.create();
+    EasyMock.expect(loggingRpcMock.write(request)).andReturn(apiFuture);
+    EasyMock.replay(rpcFactoryMock, loggingRpcMock);
+    logging = options.getService();
+    logging.write(
+        ImmutableList.of(
+            LOG_ENTRY1, LOG_ENTRY2.toBuilder().setSeverity(Severity.EMERGENCY).build()));
+  }
+
+  @Test
+  public void testWriteLogEntriesWithSeverityFlushEnabled() {
+    WriteLogEntriesRequest request =
+        WriteLogEntriesRequest.newBuilder()
+            .addAllEntries(
+                Iterables.transform(
+                    ImmutableList.of(LOG_ENTRY1, LOG_ENTRY2), LogEntry.toPbFunction(PROJECT)))
+            .build();
+    WriteLogEntriesResponse response = WriteLogEntriesResponse.newBuilder().build();
+    EasyMock.expect(loggingRpcMock.write(request)).andReturn(ApiFutures.immediateFuture(response));
+    EasyMock.replay(rpcFactoryMock, loggingRpcMock);
+    logging = options.getService();
+    logging.setFlushSeverity(Severity.DEFAULT);
+    logging.write(ImmutableList.of(LOG_ENTRY1, LOG_ENTRY2));
+  }
+
   @Test
   public void testWriteLogEntriesWithOptions() {
     Map<String, String> labels = ImmutableMap.of("key", "value");