diff --git a/server/block/register.go b/server/block/register.go
index 79f983588..e4b65c95e 100644
--- a/server/block/register.go
+++ b/server/block/register.go
@@ -305,8 +305,9 @@ func init() {
 	world.RegisterItem(Terracotta{})
 	world.RegisterItem(Tuff{})
 	world.RegisterItem(WheatSeeds{})
-	world.RegisterItem(item.Bucket{Content: Lava{}})
-	world.RegisterItem(item.Bucket{Content: Water{}})
+	world.RegisterItem(item.Bucket{Content: item.LiquidBucketContent(Lava{})})
+	world.RegisterItem(item.Bucket{Content: item.LiquidBucketContent(Water{})})
+	world.RegisterItem(item.Bucket{Content: item.MilkBucketContent()})
 
 	for _, b := range allLight() {
 		world.RegisterItem(b.(world.Item))
diff --git a/server/item/bucket.go b/server/item/bucket.go
index a702a44dd..551e18953 100644
--- a/server/item/bucket.go
+++ b/server/item/bucket.go
@@ -8,10 +8,34 @@ import (
 	"time"
 )
 
-// Bucket is a tool used to carry water, lava, milk and fish.
+// BucketContent is the content of a bucket.
+type BucketContent struct {
+	liquid world.Liquid
+	milk   bool
+}
+
+// LiquidBucketContent returns a new BucketContent with the liquid passed in.
+func LiquidBucketContent(l world.Liquid) BucketContent {
+	return BucketContent{liquid: l}
+}
+
+// MilkBucketContent returns a new BucketContent with the milk flag set.
+func MilkBucketContent() BucketContent {
+	return BucketContent{milk: true}
+}
+
+// LiquidType returns the type of liquid the bucket contains.
+func (b BucketContent) LiquidType() string {
+	if b.liquid != nil {
+		return b.liquid.LiquidType()
+	}
+	return "milk"
+}
+
+// Bucket is a tool used to carry water, lava and fish.
 type Bucket struct {
 	// Content is the content that the bucket has. By default, this value resolves to an empty bucket.
-	Content world.Liquid
+	Content BucketContent
 }
 
 // MaxCount returns 16.
@@ -22,14 +46,37 @@ func (b Bucket) MaxCount() int {
 	return 1
 }
 
+// AlwaysConsumable ...
+func (b Bucket) AlwaysConsumable() bool {
+	return b.Content.milk
+}
+
+// CanConsume ...
+func (b Bucket) CanConsume() bool {
+	return b.Content.milk
+}
+
+// ConsumeDuration ...
+func (b Bucket) ConsumeDuration() time.Duration {
+	return DefaultConsumeDuration
+}
+
+// Consume ...
+func (b Bucket) Consume(_ *world.World, c Consumer) Stack {
+	for _, effect := range c.Effects() {
+		c.RemoveEffect(effect.Type())
+	}
+	return NewStack(Bucket{}, 1)
+}
+
 // Empty returns true if the bucket is empty.
 func (b Bucket) Empty() bool {
-	return b.Content == nil
+	return b.Content.liquid == nil && !b.Content.milk
 }
 
 // FuelInfo ...
 func (b Bucket) FuelInfo() FuelInfo {
-	if b.Content.LiquidType() == "lava" {
+	if liq := b.Content.liquid; liq != nil && liq.LiquidType() == "lava" {
 		return newFuelInfo(time.Second * 1000).WithResidue(NewStack(Bucket{}, 1))
 	}
 	return FuelInfo{}
@@ -37,10 +84,13 @@ func (b Bucket) FuelInfo() FuelInfo {
 
 // UseOnBlock handles the bucket filling and emptying logic.
 func (b Bucket) UseOnBlock(pos cube.Pos, face cube.Face, _ mgl64.Vec3, w *world.World, _ User, ctx *UseContext) bool {
+	if b.Content.milk {
+		return false
+	}
 	if b.Empty() {
 		return b.fillFrom(pos, w, ctx)
 	}
-	liq := b.Content.WithDepth(8, false)
+	liq := b.Content.liquid.WithDepth(8, false)
 	if bl := w.Block(pos); canDisplace(bl, liq) || replaceableWith(bl, liq) {
 		w.SetLiquid(pos, liq)
 	} else if bl := w.Block(pos.Side(face)); canDisplace(bl, liq) || replaceableWith(bl, liq) {
@@ -49,7 +99,7 @@ func (b Bucket) UseOnBlock(pos cube.Pos, face cube.Face, _ mgl64.Vec3, w *world.
 		return false
 	}
 
-	w.PlaySound(pos.Vec3Centre(), sound.BucketEmpty{Liquid: b.Content})
+	w.PlaySound(pos.Vec3Centre(), sound.BucketEmpty{Liquid: b.Content.liquid})
 	ctx.NewItem = NewStack(Bucket{}, 1)
 	ctx.NewItemSurvivalOnly = true
 	ctx.SubtractFromCount(1)
@@ -70,7 +120,7 @@ func (b Bucket) fillFrom(pos cube.Pos, w *world.World, ctx *UseContext) bool {
 	w.SetLiquid(pos, nil)
 	w.PlaySound(pos.Vec3Centre(), sound.BucketFill{Liquid: liquid})
 
-	ctx.NewItem = NewStack(Bucket{Content: liquid}, 1)
+	ctx.NewItem = NewStack(Bucket{Content: LiquidBucketContent(liquid)}, 1)
 	ctx.NewItemSurvivalOnly = true
 	ctx.SubtractFromCount(1)
 	return true
diff --git a/server/item/item.go b/server/item/item.go
index c6f7fb89e..0451047ff 100644
--- a/server/item/item.go
+++ b/server/item/item.go
@@ -89,6 +89,11 @@ type Consumer interface {
 	// AddEffect will overwrite any effects present if the level of the effect is higher than the existing one, or
 	// if the effects' levels are equal and the new effect has a longer duration.
 	AddEffect(e effect.Effect)
+	// RemoveEffect removes any effect that might currently be active on the Consumer.
+	RemoveEffect(e effect.Type)
+	// Effects returns any effect currently applied to the Consumer. The returned effects are guaranteed not to have
+	// expired when returned.
+	Effects() []effect.Effect
 }
 
 // DefaultConsumeDuration is the default duration that consuming an item takes. Dried kelp takes half this
diff --git a/server/player/player.go b/server/player/player.go
index 062baa236..d8e4804e7 100644
--- a/server/player/player.go
+++ b/server/player/player.go
@@ -1285,6 +1285,10 @@ func (p *Player) UseItem() {
 		p.SetHeldItems(p.subtractItem(p.damageItem(i, useCtx.Damage), useCtx.CountSub), left)
 		p.addNewItem(useCtx)
 	case item.Consumable:
+		if c, ok := usable.(interface{ CanConsume() bool }); ok && !c.CanConsume() {
+			p.ReleaseItem()
+			return
+		}
 		if !usable.AlwaysConsumable() && p.GameMode().AllowsTakingDamage() && p.Food() >= 20 {
 			// The item.Consumable is not always consumable, the player is not in creative mode and the
 			// food bar is filled: The item cannot be consumed.