--- a/net/minecraft/world/level/block/LiquidBlock.java
+++ b/net/minecraft/world/level/block/LiquidBlock.java
@@ -2,6 +2,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
+import com.google.common.collect.UnmodifiableIterator;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -31,6 +32,7 @@
 import net.minecraft.world.phys.shapes.VoxelShape;
 
 public class LiquidBlock extends Block implements BucketPickup {
+
     public static final IntegerProperty LEVEL = BlockStateProperties.LEVEL;
     protected final FlowingFluid fluid;
     private final List<FluidState> stateCache;
@@ -43,17 +45,17 @@
         this.stateCache = Lists.newArrayList();
         this.stateCache.add(fluid.getSource(false));
 
-        for(int i = 1; i < 8; ++i) {
+        for (int i = 1; i < 8; ++i) {
             this.stateCache.add(fluid.getFlowing(8 - i, false));
         }
 
         this.stateCache.add(fluid.getFlowing(8, true));
-        this.registerDefaultState(this.stateDefinition.any().setValue(LEVEL, Integer.valueOf(0)));
+        this.registerDefaultState((BlockState) ((BlockState) this.stateDefinition.any()).setValue(LiquidBlock.LEVEL, 0));
     }
 
     @Override
     public VoxelShape getCollisionShape(BlockState state, BlockGetter world, BlockPos pos, CollisionContext context) {
-        return context.isAbove(STABLE_SHAPE, pos, true) && state.getValue(LEVEL) == 0 && context.canStandOnFluid(world.getFluidState(pos.above()), state.getFluidState()) ? STABLE_SHAPE : Shapes.empty();
+        return context.isAbove(LiquidBlock.STABLE_SHAPE, pos, true) && (Integer) state.getValue(LiquidBlock.LEVEL) == 0 && context.canStandOnFluid(world.getFluidState(pos.above()), state.getFluidState()) ? LiquidBlock.STABLE_SHAPE : Shapes.empty();
     }
 
     @Override
@@ -78,8 +80,9 @@
 
     @Override
     public FluidState getFluidState(BlockState state) {
-        int i = state.getValue(LEVEL);
-        return this.stateCache.get(Math.min(i, 8));
+        int i = (Integer) state.getValue(LiquidBlock.LEVEL);
+
+        return (FluidState) this.stateCache.get(Math.min(i, 8));
     }
 
     @Override
@@ -105,11 +108,31 @@
     @Override
     public void onPlace(BlockState state, Level world, BlockPos pos, BlockState oldState, boolean notify) {
         if (this.shouldSpreadLiquid(world, pos, state)) {
-            world.scheduleTick(pos, state.getFluidState().getType(), this.fluid.getTickDelay(world));
+            world.scheduleTick(pos, state.getFluidState().getType(), this.getFlowSpeed(world, pos)); // Paper
         }
 
     }
 
+    // Paper start - Get flow speed. Throttle if its water and flowing adjacent to lava
+    public int getFlowSpeed(Level world, BlockPos blockposition) {
+        if (net.minecraft.core.registries.BuiltInRegistries.FLUID.wrapAsHolder(this.fluid).is(FluidTags.WATER)) {
+            if (
+                isLava(world, blockposition.north(1)) ||
+                isLava(world, blockposition.south(1)) ||
+                isLava(world, blockposition.west(1)) ||
+                isLava(world, blockposition.east(1))
+            ) {
+                return world.paperConfig().environment.waterOverLavaFlowSpeed;
+            }
+        }
+        return this.fluid.getTickDelay(world);
+    }
+    private static boolean isLava(Level world, BlockPos blockPos) {
+        final FluidState fluidState = world.getFluidIfLoaded(blockPos);
+        return fluidState != null && fluidState.is(FluidTags.LAVA);
+    }
+    // Paper end
+
     @Override
     public BlockState updateShape(BlockState state, Direction direction, BlockState neighborState, LevelAccessor world, BlockPos pos, BlockPos neighborPos) {
         if (state.getFluidState().isSource() || neighborState.getFluidState().isSource()) {
@@ -122,27 +145,37 @@
     @Override
     public void neighborChanged(BlockState state, Level world, BlockPos pos, Block sourceBlock, BlockPos sourcePos, boolean notify) {
         if (this.shouldSpreadLiquid(world, pos, state)) {
-            world.scheduleTick(pos, state.getFluidState().getType(), this.fluid.getTickDelay(world));
+            world.scheduleTick(pos, state.getFluidState().getType(), this.getFlowSpeed(world, pos)); // Paper
         }
 
     }
 
     private boolean shouldSpreadLiquid(Level world, BlockPos pos, BlockState state) {
         if (this.fluid.is(FluidTags.LAVA)) {
-            boolean bl = world.getBlockState(pos.below()).is(Blocks.SOUL_SOIL);
+            boolean flag = world.getBlockState(pos.below()).is(Blocks.SOUL_SOIL);
+            UnmodifiableIterator unmodifiableiterator = LiquidBlock.POSSIBLE_FLOW_DIRECTIONS.iterator();
 
-            for(Direction direction : POSSIBLE_FLOW_DIRECTIONS) {
-                BlockPos blockPos = pos.relative(direction.getOpposite());
-                if (world.getFluidState(blockPos).is(FluidTags.WATER)) {
+            while (unmodifiableiterator.hasNext()) {
+                Direction enumdirection = (Direction) unmodifiableiterator.next();
+                BlockPos blockposition1 = pos.relative(enumdirection.getOpposite());
+
+                if (world.getFluidState(blockposition1).is(FluidTags.WATER)) {
                     Block block = world.getFluidState(pos).isSource() ? Blocks.OBSIDIAN : Blocks.COBBLESTONE;
-                    world.setBlockAndUpdate(pos, block.defaultBlockState());
-                    this.fizz(world, pos);
+
+                    // CraftBukkit start
+                    if (org.bukkit.craftbukkit.v1_20_R2.event.CraftEventFactory.handleBlockFormEvent(world, pos, block.defaultBlockState())) {
+                        this.fizz(world, pos);
+                    }
+                    // CraftBukkit end
                     return false;
                 }
 
-                if (bl && world.getBlockState(blockPos).is(Blocks.BLUE_ICE)) {
-                    world.setBlockAndUpdate(pos, Blocks.BASALT.defaultBlockState());
-                    this.fizz(world, pos);
+                if (flag && world.getBlockState(blockposition1).is(Blocks.BLUE_ICE)) {
+                    // CraftBukkit start
+                    if (org.bukkit.craftbukkit.v1_20_R2.event.CraftEventFactory.handleBlockFormEvent(world, pos, Blocks.BASALT.defaultBlockState())) {
+                        this.fizz(world, pos);
+                    }
+                    // CraftBukkit end
                     return false;
                 }
             }
@@ -157,12 +190,12 @@
 
     @Override
     protected void createBlockStateDefinition(StateDefinition.Builder<Block, BlockState> builder) {
-        builder.add(LEVEL);
+        builder.add(LiquidBlock.LEVEL);
     }
 
     @Override
     public ItemStack pickupBlock(@Nullable Player player, LevelAccessor world, BlockPos pos, BlockState state) {
-        if (state.getValue(LEVEL) == 0) {
+        if ((Integer) state.getValue(LiquidBlock.LEVEL) == 0) {
             world.setBlock(pos, Blocks.AIR.defaultBlockState(), 11);
             return new ItemStack(this.fluid.getBucket());
         } else {
