--- 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 com.mojang.serialization.Codec;
 import com.mojang.serialization.DataResult;
 import com.mojang.serialization.MapCodec;
@@ -28,7 +29,6 @@
 import net.minecraft.world.level.block.state.properties.BlockStateProperties;
 import net.minecraft.world.level.block.state.properties.IntegerProperty;
 import net.minecraft.world.level.material.FlowingFluid;
-import net.minecraft.world.level.material.Fluid;
 import net.minecraft.world.level.material.FluidState;
 import net.minecraft.world.level.pathfinder.PathComputationType;
 import net.minecraft.world.level.storage.loot.LootParams;
@@ -37,26 +37,36 @@
 import net.minecraft.world.phys.shapes.VoxelShape;
 
 public class LiquidBlock extends Block implements BucketPickup {
-    private static final Codec<FlowingFluid> FLOWING_FLUID = BuiltInRegistries.FLUID
-        .byNameCodec()
-        .comapFlatMap(
-            fluid -> fluid instanceof FlowingFluid flowingFluid ? DataResult.success(flowingFluid) : DataResult.error(() -> "Not a flowing fluid: " + fluid),
-            fluid -> (Fluid)fluid
-        );
-    public static final MapCodec<LiquidBlock> CODEC = RecordCodecBuilder.mapCodec(
-        instance -> instance.group(FLOWING_FLUID.fieldOf("fluid").forGetter(block -> block.fluid), propertiesCodec()).apply(instance, LiquidBlock::new)
-    );
+
+    private static final Codec<FlowingFluid> FLOWING_FLUID = BuiltInRegistries.FLUID.byNameCodec().comapFlatMap((fluidtype) -> {
+        DataResult<FlowingFluid> dataresult; // CraftBukkit - decompile error
+
+        if (fluidtype instanceof FlowingFluid fluidtypeflowing) {
+            dataresult = DataResult.success(fluidtypeflowing);
+        } else {
+            dataresult = DataResult.error(() -> {
+                return "Not a flowing fluid: " + String.valueOf(fluidtype);
+            });
+        }
+
+        return dataresult;
+    }, (fluidtypeflowing) -> {
+        return fluidtypeflowing;
+    });
+    public static final MapCodec<LiquidBlock> CODEC = RecordCodecBuilder.mapCodec((instance) -> {
+        return instance.group(LiquidBlock.FLOWING_FLUID.fieldOf("fluid").forGetter((blockfluids) -> {
+            return blockfluids.fluid;
+        }), propertiesCodec()).apply(instance, LiquidBlock::new);
+    });
     public static final IntegerProperty LEVEL = BlockStateProperties.LEVEL;
     protected final FlowingFluid fluid;
     private final List<FluidState> stateCache;
-    public static final VoxelShape STABLE_SHAPE = Block.box(0.0, 0.0, 0.0, 16.0, 8.0, 16.0);
-    public static final ImmutableList<Direction> POSSIBLE_FLOW_DIRECTIONS = ImmutableList.of(
-        Direction.DOWN, Direction.SOUTH, Direction.NORTH, Direction.EAST, Direction.WEST
-    );
+    public static final VoxelShape STABLE_SHAPE = Block.box(0.0D, 0.0D, 0.0D, 16.0D, 8.0D, 16.0D);
+    public static final ImmutableList<Direction> POSSIBLE_FLOW_DIRECTIONS = ImmutableList.of(Direction.DOWN, Direction.SOUTH, Direction.NORTH, Direction.EAST, Direction.WEST);
 
     @Override
     public MapCodec<LiquidBlock> codec() {
-        return CODEC;
+        return LiquidBlock.CODEC;
     }
 
     protected LiquidBlock(FlowingFluid fluid, BlockBehaviour.Properties settings) {
@@ -65,21 +75,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
     protected 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
@@ -104,8 +110,9 @@
 
     @Override
     protected 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
@@ -131,9 +138,30 @@
     @Override
     protected 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 - Configurable speed for water flowing over lava
         }
+
+    }
+
+    // Paper start - Configurable speed for water flowing over 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 - Configurable speed for water flowing over lava
 
     @Override
     protected BlockState updateShape(BlockState state, Direction direction, BlockState neighborState, LevelAccessor world, BlockPos pos, BlockPos neighborPos) {
@@ -147,26 +175,37 @@
     @Override
     protected 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 - Configurable speed for water flowing over lava
         }
+
     }
 
     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.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.event.CraftEventFactory.handleBlockFormEvent(world, pos, Blocks.BASALT.defaultBlockState())) {
+                        this.fizz(world, pos);
+                    }
+                    // CraftBukkit end
                     return false;
                 }
             }
@@ -181,12 +220,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 {
