diff --git a/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java b/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java
index b5a93e68942..2b7e3625896 100644
--- a/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java
+++ b/server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java
@@ -313,10 +313,12 @@ public class RealtimePlumber implements Plumber
             final Interval interval = sink.getInterval();
 
             for (FireHydrant hydrant : sink) {
-              if (!hydrant.hasSwapped()) {
-                log.info("Hydrant[%s] hasn't swapped yet, swapping. Sink[%s]", hydrant, sink);
-                final int rowCount = persistHydrant(hydrant, schema, interval);
-                metrics.incrementRowOutputCount(rowCount);
+              synchronized (hydrant) {
+                if (!hydrant.hasSwapped()) {
+                  log.info("Hydrant[%s] hasn't swapped yet, swapping. Sink[%s]", hydrant, sink);
+                  final int rowCount = persistHydrant(hydrant, schema, interval);
+                  metrics.incrementRowOutputCount(rowCount);
+                }
               }
             }