Real-Time Minecraft Telemetry Pipeline

Introduction

30 Second Summary

Multiplayer game servers generate a firehose of data every second. Player joins, combat events, performance metrics. The challenge is capturing all of it without slowing down the game, then making sense of it in real time.

In this project, you will build a complete telemetry pipeline that streams live data from a Minecraft server into a centralized analytics dashboard updating every 500 milliseconds.

What You'll Build

A production-grade real-time data pipeline using Valkey Streams, Protocol Buffers, and WebSocket push to move thousands of events per second from game servers to a live dashboard.

By the end of this project, you'll have:

• A Paper plugin written in Kotlin that collects player events and server metrics completely off the main thread and publishes them as Protobuf-encoded messages to Valkey Streams.
• A Ktor analytics service that consumes from Valkey via consumer groups and pushes aggregated data to connected clients over WebSocket.
• A real-time HTML dashboard with Chart.js showing per-server player counts, TPS, and event throughput at sub-second intervals.
• Secret Mission: Add per-server event breakdown charts and historical sparklines to the dashboard.

Are there any prerequisites?

This project assumes you have experience writing Kotlin plugins for Paper servers. You should also have Docker Desktop installed and a working Paper server environment.

You do not need prior experience with Valkey, Ktor, or Protocol Buffers. The project introduces each of these from scratch.

Set Up Redis and Project Structure

Set Up Valkey and Project Structure

Every telemetry pipeline starts with a message broker sitting between producers and consumers. In this project, that broker is Valkey Streams.

Before you write any telemetry code, you need three things in place: a running Valkey instance to receive events, a Ktor service project to consume them, and a Paper plugin module to produce them.

In this step, get ready to:

• Start a Valkey container with Docker and verify the connection.
• Create the Ktor analytics service project with all required dependencies.
• Create the Paper plugin module with its dependencies and the shared TelemetryEvent data class.

Start Valkey via Docker

Valkey will act as the central nervous system of your pipeline. Every player event and server metric flows through it as a stream entry before the analytics service picks it up.

• Open a terminal window by pressing Cmd+Space (macOS) or the Windows key (Windows), then typing Terminal and pressing Enter.
• Start a Valkey container by running this command:

docker run -d --name valkey-telemetry -p 6379:6379 valkey/valkey:latest

What does this command do?

• -d runs the container in detached mode (in the background).
• --name valkey-telemetry gives the container a memorable name you can reference later.
• -p 6379:6379 maps the container's Valkey port to your machine's port 6379.
• valkey/valkey:latest pulls the official Valkey image.

• Verify Valkey is running by pinging it:

docker exec valkey-telemetry valkey-cli PING

• You should see PONG printed in your terminal. This confirms Valkey is accepting connections.

Not seeing PONG?

• Make sure Docker Desktop is running. If it is not, start it and try the docker run command again.
• If you see a "name already in use" error, the container already exists. Run docker rm -f valkey-telemetry and then re-run the docker run command.
• If port 6379 is already in use, stop whatever process is using it or change the host port (e.g. -p 6380:6379).

Still stuck? Help me troubleshoot my Valkey Docker container.

Create the Ktor analytics service project

The analytics service is a standalone Kotlin application that will consume telemetry from Valkey Streams and push aggregated data to a live dashboard via WebSocket. You will scaffold it as a Gradle Kotlin DSL project in IntelliJ IDEA.

• Switch back to IntelliJ IDEA.
• Click File in the top menu bar.
• Click New, then Project....
• Set the project name to ktor-analytics-service.
• Select Kotlin as the language.
• Select Gradle as the build system with Kotlin DSL.
• Click Create.

IntelliJ creates the project structure with a default build.gradle.kts file.

• Open build.gradle.kts in the Project tool window on the left.
• Replace its entire contents with the following:

plugins {
    kotlin("jvm") version "2.3.21"
    kotlin("plugin.serialization") version "2.3.21"
}

group = "com.telemetry"
version = "1.0.0"

repositories {
    mavenCentral()
}

dependencies {
    // Ktor server framework
    implementation("io.ktor:ktor-server-core:3.5.1")
    implementation("io.ktor:ktor-server-netty:3.5.1")
    implementation("io.ktor:ktor-server-websockets:3.5.1")

    // Valkey client for Streams (Lettuce speaks the same protocol)
    implementation("io.lettuce:lettuce-core:7.6.0.RELEASE")

    // Protobuf serialization for telemetry envelopes
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-protobuf:1.10.0")
}

What does this build file set up?

• ktor-server-core, ktor-server-netty, and ktor-server-websockets give you an HTTP server with WebSocket support.
• lettuce-core is a high-performance client that speaks the Valkey protocol and supports coroutines natively.
• kotlinx-serialization-protobuf encodes and decodes your telemetry data as compact Protocol Buffer bytes.
• The plugin.serialization plugin generates serializers at compile time for any class you annotate with @Serializable.

• Save the file.
• Click the Sync icon in the notification bar that appears (or press Cmd+Shift+I on macOS, Ctrl+Shift+O on Windows) to let Gradle download the dependencies.
• Wait for the sync to complete. You should see a green checkmark in the Build tool window at the bottom.

Gradle sync failed?

• Check that your JDK version is 17 or higher. Go to File > Project Structure > SDK and verify.
• Make sure you have an internet connection. Gradle needs to download dependencies from Maven Central.
• If you see version resolution errors, double-check that the dependency strings match exactly what is shown above.

Need more help? Help me fix my Gradle sync error in IntelliJ IDEA.

Create the Paper plugin module and shared TelemetryEvent

The Paper plugin is the producer side of your pipeline. It runs inside the Minecraft server and collects events. Both the plugin and the analytics service need to understand the same message format, so you will also define a shared Protocol Buffers envelope class called TelemetryEvent.

• In IntelliJ IDEA, click File > New > Module....
• Set the module name to paper-plugin.
• Select Kotlin as the language and Gradle with Kotlin DSL as the build system.
• Click Create.

• Open the build.gradle.kts file inside the paper-plugin module.
• Replace its contents with:

plugins {
    kotlin("jvm") version "2.3.21"
    kotlin("plugin.serialization") version "2.3.21"
}

group = "com.telemetry"
version = "1.0.0"

repositories {
    mavenCentral()
    maven("https://repo.papermc.io/repository/maven-public/")
}

dependencies {
    // Paper server API (provided at runtime by the server)
    compileOnly("io.papermc.paper:paper-api:26.2.build.+")

    // Valkey client for publishing to Streams (Lettuce speaks the same protocol)
    implementation("io.lettuce:lettuce-core:7.6.0.RELEASE")

    // Protobuf serialization for telemetry envelopes
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-protobuf:1.10.0")
}

Why is paper-api marked compileOnly?

The Paper server already provides the API classes at runtime. Marking the dependency as compileOnly means Gradle uses it during compilation but does not bundle it into your plugin JAR. This keeps the plugin lightweight.

• Save the file and sync Gradle again.

Now create the shared data class that both modules use.

• In the Project tool window, right-click the paper-plugin/src/main/kotlin folder.
• Select New > Package and name it com.telemetry.shared.
• Right-click the new package and select New > Kotlin Class/File.
• Name it TelemetryEvent and select File.
• Paste the following code into TelemetryEvent.kt:

package com.telemetry.shared

import kotlinx.serialization.ExperimentalSerializationApi
import kotlinx.serialization.Serializable

// The shared envelope for all telemetry messages
@OptIn(ExperimentalSerializationApi::class)
@Serializable
data class TelemetryEvent(
    // Event category: "player_join", "player_quit", "entity_death", "server_tick"
    val type: String,
    // Identifies which Minecraft server sent this event
    val serverId: String,
    // Unix epoch millis when the event occurred
    val timestamp: Long,
    // Human-readable event data (maps to protobuf string type)
    val payload: String
)

Why use String for the payload?

The envelope pattern lets you forward heterogeneous Minecraft events without creating a data class for every packet type. The payload field holds a human-readable string that the consumer can parse directly without extra decoding steps.

Using String keeps the pipeline simple. The producer formats data as a delimited string, and the consumer splits on the delimiter. No byte-to-string conversion needed on either side.

• Save TelemetryEvent.kt.
• Confirm the file has no red underlines in the editor. If IntelliJ shows unresolved references, re-sync Gradle.

Seeing unresolved reference errors?

• Make sure the plugin.serialization plugin is listed in your build.gradle.kts plugins block. Without it, @Serializable will not resolve.
• Verify the kotlinx-serialization-protobuf dependency is present in the dependencies block.
• Try File > Invalidate Caches / Restart if re-syncing Gradle does not fix it.

Still seeing errors? Help me resolve unresolved reference errors for kotlinx.serialization.

✔️ Awesome, I've got everything!

Great. Double check you have saved all files and that Gradle sync completed with no errors.

ⓧ I'd like to double check the full code

Here is the full build.gradle.kts for the Ktor analytics service:

plugins {
    kotlin("jvm") version "2.3.21"
    kotlin("plugin.serialization") version "2.3.21"
}

group = "com.telemetry"
version = "1.0.0"

repositories {
    mavenCentral()
}

dependencies {
    // Ktor server framework
    implementation("io.ktor:ktor-server-core:3.5.1")
    implementation("io.ktor:ktor-server-netty:3.5.1")
    implementation("io.ktor:ktor-server-websockets:3.5.1")

    // Valkey client for Streams (Lettuce speaks the same protocol)
    implementation("io.lettuce:lettuce-core:7.6.0.RELEASE")

    // Protobuf serialization for telemetry envelopes
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-protobuf:1.10.0")
}

Here is the full build.gradle.kts for the Paper plugin module:

plugins {
    kotlin("jvm") version "2.3.21"
    kotlin("plugin.serialization") version "2.3.21"
}

group = "com.telemetry"
version = "1.0.0"

repositories {
    mavenCentral()
    maven("https://repo.papermc.io/repository/maven-public/")
}

dependencies {
    // Paper server API (provided at runtime by the server)
    compileOnly("io.papermc.paper:paper-api:26.2.build.+")

    // Valkey client for publishing to Streams (Lettuce speaks the same protocol)
    implementation("io.lettuce:lettuce-core:7.6.0.RELEASE")

    // Protobuf serialization for telemetry envelopes
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-protobuf:1.10.0")
}

And here is the shared TelemetryEvent.kt:

package com.telemetry.shared

import kotlinx.serialization.ExperimentalSerializationApi
import kotlinx.serialization.Serializable

// The shared envelope for all telemetry messages
@OptIn(ExperimentalSerializationApi::class)
@Serializable
data class TelemetryEvent(
    // Event category: "player_join", "player_quit", "entity_death", "server_tick"
    val type: String,
    // Identifies which Minecraft server sent this event
    val serverId: String,
    // Unix epoch millis when the event occurred
    val timestamp: Long,
    // Human-readable event data (maps to protobuf string type)
    val payload: String
)

Your infrastructure is in place. You have Valkey accepting connections, a Ktor project ready for stream consumption, and a Paper plugin module with a shared message envelope. Next up, you will build the plugin that collects player events and publishes them to Valkey Streams.

Build the Paper Plugin Telemetry Producer

Your project structure and Redis instance are ready. Now you need to build the plugin that actually captures what happens on your Minecraft server and streams it out.

The challenge is doing this without affecting server performance. Minecraft servers run on a single main thread at 20 ticks per second. In this step, you will build a packet interception layer using PacketEvents that captures player and protocol-level data directly on the Netty I/O thread, then queues it for publishing to Valkey Streams entirely off the main thread.

In this step, get ready to:

• Build a PacketEvents listener that dynamically captures every inbound and outbound packet into a lock-free queue.
• Build a background flush loop that drains the queue and publishes Protobuf-encoded events to Redis every 250ms.
• Add a TPS/MSPT sampler that feeds server health metrics into the same pipeline.

Implement the packet telemetry listener

PacketEvents intercepts every raw Minecraft protocol packet on the Netty I/O thread. The naive approach is to use Method.invoke() reflection on every packet, but that causes CPU spikes because the JVM cannot inline reflective calls. Instead, you will use MethodHandles to compile getters once per packet type on first encounter, then reuse them at near-native speed on the hot path.

A ConcurrentLinkedQueue bridges the Netty threads and the flush loop. The extractorCache stores a FieldExtractor per wrapper class containing the pre-compiled constructor and getter handles. After the first encounter of each packet type, all subsequent extractions run without reflection.

• Create a new file called PacketTelemetryListener.kt inside paper-plugin/src/main/kotlin/ and paste the following code:

package com.example.telemetry

import com.github.retrooper.packetevents.event.PacketListener
import com.github.retrooper.packetevents.event.PacketReceiveEvent
import com.github.retrooper.packetevents.event.PacketSendEvent
import com.github.retrooper.packetevents.event.UserDisconnectEvent
import com.github.retrooper.packetevents.event.UserLoginEvent
import com.github.retrooper.packetevents.protocol.packettype.PacketTypeCommon
import com.github.retrooper.packetevents.wrapper.PacketWrapper
import java.lang.invoke.MethodHandle
import java.lang.invoke.MethodHandles
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue

// Holds pre-compiled MethodHandles for a single wrapper class (built once, zero-cost reuse)
data class FieldExtractor(
    val constructorHandle: MethodHandle,
    val getters: List<Pair<String, MethodHandle>>
)

// Intercepts ALL network packets and extracts fields via MethodHandles (no hot-path reflection)
class PacketTelemetryListener(
    private val queue: ConcurrentLinkedQueue<TelemetryEvent>,
    private val serverId: String
) : PacketListener {

    private val lookup = MethodHandles.lookup()

    // Cache of pre-compiled extractors per wrapper class (built once, reused forever)
    private val extractorCache = ConcurrentHashMap<Class<*>, FieldExtractor?>()

    // Methods inherited from PacketWrapper internals that don't carry useful telemetry
    private val ignoredMethods = setOf(
        "getBuffer", "getServerVersion", "getClientVersion",
        "getPacketId", "getNativePacketId", "getPacketType",
        "getClass", "getChannel"
    )

    override fun onUserLogin(event: UserLoginEvent) {
        val user = event.getUser()
        val telemetryEvent = TelemetryEvent(
            type = "player_join",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = user.name
        )
        queue.add(telemetryEvent)
    }

    override fun onUserDisconnect(event: UserDisconnectEvent) {
        val user = event.getUser()
        val telemetryEvent = TelemetryEvent(
            type = "player_quit",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = user.name
        )
        queue.add(telemetryEvent)
    }

    override fun onPacketReceive(event: PacketReceiveEvent) {
        val packetName = event.packetType.name
        val userName = event.getUser().name
        // Extract fields using pre-compiled MethodHandles (no per-packet reflection)
        val fields = extractFields(event.packetType, event)
        val telemetryEvent = TelemetryEvent(
            type = "packet_in:$packetName",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = "$userName|$fields"
        )
        queue.add(telemetryEvent)
    }

    override fun onPacketSend(event: PacketSendEvent) {
        val packetName = event.packetType.name
        val userName = event.getUser().name
        // Extract fields using pre-compiled MethodHandles (no per-packet reflection)
        val fields = extractFields(event.packetType, event)
        val telemetryEvent = TelemetryEvent(
            type = "packet_out:$packetName",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = "$userName|$fields"
        )
        queue.add(telemetryEvent)
    }

    private fun extractFields(packetType: PacketTypeCommon, event: Any): String {
        val wrapperClass = packetType.wrapperClass ?: return "raw"
        // Get or build the extractor (reflection runs only on first encounter)
        val extractor = extractorCache.getOrPut(wrapperClass) { buildExtractor(wrapperClass, event) }
            ?: return "raw"
        return try {
            // Instantiate wrapper via pre-compiled constructor handle
            val wrapper = extractor.constructorHandle.invoke(event)
            // Call each getter via pre-compiled method handles
            val fields = extractor.getters.mapNotNull { (name, handle) ->
                try {
                    val value = handle.invoke(wrapper)
                    "$name=${valueToString(value)}"
                } catch (_: Throwable) {
                    null
                }
            }
            fields.joinToString(",")
        } catch (_: Throwable) {
            "raw"
        }
    }

    // Runs ONCE per packet type — uses reflection to discover, then compiles to MethodHandles
    private fun buildExtractor(wrapperClass: Class<*>, event: Any): FieldExtractor? {
        return try {
            // Find and compile the constructor into a MethodHandle
            val constructor = wrapperClass.getConstructor(event::class.java)
            val constructorHandle = lookup.unreflectConstructor(constructor)
            // Find getter methods, filter internals, compile each to a MethodHandle
            val getters = wrapperClass.methods
                .filter { it.name.startsWith("get") }
                .filter { it.parameterCount == 0 }
                .filter { it.name !in ignoredMethods }
                .filter { !PacketWrapper::class.java.methods.any { base -> base.name == it.name } || isOverridden(it, wrapperClass) }
                .sortedBy { it.name }
                .map { method ->
                    val key = method.name.removePrefix("get").replaceFirstChar { it.lowercase() }
                    // Compile the Method into a MethodHandle (JIT-inlinable after warmup)
                    val handle = lookup.unreflect(method)
                    Pair(key, handle)
                }
            FieldExtractor(constructorHandle, getters)
        } catch (_: Exception) {
            null
        }
    }

    private fun isOverridden(method: java.lang.reflect.Method, wrapperClass: Class<*>): Boolean {
        // Check if the method is declared in the wrapper subclass (not just inherited)
        return try {
            wrapperClass.getDeclaredMethod(method.name, *method.parameterTypes)
            true
        } catch (_: NoSuchMethodException) {
            false
        }
    }

    private fun valueToString(value: Any?): String {
        return when (value) {
            null -> "null"
            is ByteArray -> "[${value.size} bytes]"
            is Collection<*> -> "[${value.size} items]"
            is Array<*> -> "[${value.size} items]"
            else -> value.toString().take(128)
        }
    }
}

What does this code do?

The buildExtractor function runs once per packet type on its first encounter. It uses reflection to discover the wrapper constructor and all get* methods, then compiles each one into a MethodHandle via lookup.unreflect().

After JIT warmup, MethodHandle.invoke() compiles down to a direct method call. No security checks, no argument boxing, no Method.invoke() overhead. The hot path (every packet) only calls pre-compiled handles stored in the extractorCache.

The FieldExtractor data class bundles the constructor handle with the getter handles for each wrapper class. Once the cache is warm (after seeing each packet type once), the entire extraction pipeline runs without any reflective calls.

You still get full dynamic extraction of all fields from all packet types. New packets added by Mojang still work automatically. The difference is purely in how the JVM executes the calls at runtime.

• Save PacketTelemetryListener.kt.

You will not see a change yet. The next chunk builds the flush loop that actually sends these queued events to Valkey.

Implement the background flush loop

Events are piling up in the queue. You need a background task that wakes up every 250ms, drains everything in the queue, encodes each event with Protocol Buffers, and fires them into Valkey Streams using Lettuce's async API.

• Create a new file called FlushLoop.kt inside paper-plugin/src/main/kotlin/ and paste the following code:

package com.example.telemetry

import io.lettuce.core.XAddArgs
import io.lettuce.core.api.StatefulRedisConnection
import kotlinx.serialization.ExperimentalSerializationApi
import kotlinx.serialization.protobuf.ProtoBuf
import org.bukkit.plugin.java.JavaPlugin
import java.util.Base64
import java.util.concurrent.ConcurrentLinkedQueue

// Drains the event queue on a background thread and publishes to Valkey Streams
class FlushLoop(
    private val plugin: JavaPlugin,
    private val queue: ConcurrentLinkedQueue<TelemetryEvent>,
    private val connection: StatefulRedisConnection<String, String>,
    private val serverId: String
) {

    @OptIn(ExperimentalSerializationApi::class)
    fun start() {
        // Schedule an async repeating task: 0 tick delay, 5 tick period (250ms at 20 TPS)
        plugin.server.scheduler.runTaskTimerAsynchronously(plugin, Runnable {
            val commands = connection.async()
            // Drain all queued events in one pass
            while (true) {
                val event = queue.poll() ?: break
                // Serialize the TelemetryEvent to Protobuf bytes
                val encoded = ProtoBuf.encodeToByteArray(TelemetryEvent.serializer(), event)
                // Base64-encode for safe storage in a String-codec connection
                val base64 = Base64.getEncoder().encodeToString(encoded)
                // Publish to the server-specific stream with bounded retention
                val args = XAddArgs.Builder.maxlen(10000).approximateTrimming()
                commands.xadd("telemetry:$serverId", args, mapOf("data" to base64))
            }
        }, 0L, 5L)
    }
}

What does this code do?

The runTaskTimerAsynchronously method schedules a repeating task that runs off the main server thread. With a period of 5 ticks at 20 TPS, it fires every 250ms.

Inside the task, queue.poll() drains events one at a time (returns null when empty). Each event is Protobuf-serialized with ProtoBuf.encodeToByteArray(), then Base64-encoded so it fits cleanly into Lettuce's String codec.

The XAddArgs.Builder.maxlen(10000).approximateTrimming() caps each stream at roughly 10,000 entries. The approximate flag (MAXLEN ~ in Redis) lets Redis trim at macro-node boundaries, which is far cheaper than exact trimming on a busy stream.

Why Protobuf for serialization?

The kotlinx-serialization-protobuf module is officially maintained by JetBrains as part of kotlinx.serialization. It maps Kotlin String directly to the native protobuf string wire type. The payload stays human-readable through the entire pipeline without extra encoding steps.

• Save FlushLoop.kt.

You will not see output yet. The flush loop needs the main plugin class to start it, which you will wire up after the TPS sampler.

Add a TPS/MSPT sampler

Player events alone do not tell the full story. The analytics dashboard also needs to know how healthy each server is. TPS (ticks per second) indicates overall capacity. MSPT (milliseconds per tick) reveals how much headroom remains before lag hits.

• Create a new file called TpsSampler.kt inside paper-plugin/src/main/kotlin/ and paste the following code:

package com.example.telemetry

import org.bukkit.Bukkit
import org.bukkit.plugin.java.JavaPlugin
import java.util.concurrent.ConcurrentLinkedQueue

// Samples server TPS and MSPT every 500ms and enqueues as telemetry events
class TpsSampler(
    private val plugin: JavaPlugin,
    private val queue: ConcurrentLinkedQueue<TelemetryEvent>,
    private val serverId: String
) {

    fun start() {
        // Schedule an async repeating task: 0 tick delay, 10 tick period (500ms at 20 TPS)
        plugin.server.scheduler.runTaskTimerAsynchronously(plugin, Runnable {
            // Read server performance metrics
            val tps = Bukkit.getTPS()[0]
            val mspt = Bukkit.getAverageTickTime()
            // Pack TPS and MSPT into the payload as a delimited string
            val event = TelemetryEvent(
                type = "server_tick",
                serverId = serverId,
                timestamp = System.currentTimeMillis(),
                payload = "$tps:$mspt"
            )
            queue.add(event)
        }, 0L, 10L)
    }
}

What does this code do?

This sampler runs every 10 ticks (500ms). It reads Bukkit.getTPS()[0] for the most recent 1-minute TPS average and Bukkit.getAverageTickTime() for the average milliseconds per tick.

The values are packed into a simple "tps:mspt" string payload. The consumer service will parse this on the other side. Using the same TelemetryEvent envelope with type = "server_tick" means the entire pipeline handles it identically to player events.

• Save TpsSampler.kt.

Now you have all three components. Time to wire them together in the main plugin class.

• Create a new file called TelemetryPlugin.kt inside paper-plugin/src/main/kotlin/ and paste the following code:

package com.example.telemetry

import com.github.retrooper.packetevents.PacketEvents
import com.github.retrooper.packetevents.event.PacketListenerPriority
import io.github.retrooper.packetevents.factory.spigot.SpigotPacketEventsBuilder
import io.lettuce.core.RedisClient
import io.lettuce.core.api.StatefulRedisConnection
import org.bukkit.plugin.java.JavaPlugin
import java.util.concurrent.ConcurrentLinkedQueue

// Main plugin class that initializes PacketEvents, Valkey, and all telemetry components
class TelemetryPlugin : JavaPlugin() {

    private lateinit var redisClient: RedisClient
    private lateinit var connection: StatefulRedisConnection<String, String>
    private val eventQueue = ConcurrentLinkedQueue<TelemetryEvent>()
    private val serverId = "server-1"

    override fun onLoad() {
        // Build and load the PacketEvents API instance
        PacketEvents.setAPI(SpigotPacketEventsBuilder.build(this))
        PacketEvents.getAPI().load()
    }

    override fun onEnable() {
        // Register the packet telemetry listener before initializing
        val listener = PacketTelemetryListener(eventQueue, serverId)
        PacketEvents.getAPI().eventManager.registerListener(listener, PacketListenerPriority.NORMAL)

        // Initialize PacketEvents (must happen after listener registration)
        PacketEvents.getAPI().init()

        // Initialize Lettuce client connecting to Valkey on localhost
        redisClient = RedisClient.create("redis://localhost:6379")
        connection = redisClient.connect()

        // Start the background flush loop (drains queue every 250ms)
        val flushLoop = FlushLoop(this, eventQueue, connection, serverId)
        flushLoop.start()

        // Start the TPS/MSPT sampler (samples every 500ms)
        val tpsSampler = TpsSampler(this, eventQueue, serverId)
        tpsSampler.start()

        logger.info("TelemetryPlugin enabled - streaming to Valkey via PacketEvents")
    }

    override fun onDisable() {
        // Terminate PacketEvents cleanly
        PacketEvents.getAPI().terminate()
        // Clean up Valkey connection on plugin disable
        connection.close()
        redisClient.shutdown()
        logger.info("TelemetryPlugin disabled - Valkey connection closed")
    }
}

What does this code do?

The TelemetryPlugin class manages the PacketEvents lifecycle across three hooks. In onLoad(), it builds and loads the PacketEvents API using SpigotPacketEventsBuilder. In onEnable(), it registers the listener before calling init(). This ordering is required by PacketEvents.

It shares a single ConcurrentLinkedQueue across the PacketTelemetryListener, FlushLoop, and TpsSampler. All telemetry data flows through a single queue into a single flush path. On disable, it calls terminate() to shut down PacketEvents, then closes the Valkey connection.

• Save TelemetryPlugin.kt.
• Verify the project compiles by opening the IntelliJ terminal and running a Gradle build:

./gradlew :paper-plugin:build

You should see BUILD SUCCESSFUL in the terminal output. The compiled plugin JAR is ready for deployment to your Paper server.

Seeing a compile error?

• Check that your TelemetryEvent data class from Step 1 has the @Serializable annotation and is in a package that all four files can import.
• Verify that com.github.retrooper:packetevents-spigot:2.13.0, io.lettuce:lettuce-core:7.6.0.RELEASE, and org.jetbrains.kotlinx:kotlinx-serialization-protobuf:1.10.0 are listed in your paper-plugin/build.gradle.kts dependencies.
• Make sure you have the Kotlin serialization Gradle plugin applied (kotlin("plugin.serialization") in your plugins block).
• If you see "Unresolved reference: PacketEvents", check that the packetevents-spigot dependency uses implementation (not compileOnly) since the plugin bundles it.

help me fix my Paper telemetry plugin compile error

✔️ Awesome, I've got everything!

Great. Double check you have saved all four files in the paper-plugin/src/main/kotlin/ directory.

ⓧ I'd like to double check the full code

Here are all four plugin source files together for comparison.

package com.example.telemetry

import com.github.retrooper.packetevents.event.PacketListener
import com.github.retrooper.packetevents.event.PacketReceiveEvent
import com.github.retrooper.packetevents.event.PacketSendEvent
import com.github.retrooper.packetevents.event.UserDisconnectEvent
import com.github.retrooper.packetevents.event.UserLoginEvent
import com.github.retrooper.packetevents.protocol.packettype.PacketTypeCommon
import com.github.retrooper.packetevents.wrapper.PacketWrapper
import java.lang.invoke.MethodHandle
import java.lang.invoke.MethodHandles
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue

// Holds pre-compiled MethodHandles for a single wrapper class (built once, zero-cost reuse)
data class FieldExtractor(
    val constructorHandle: MethodHandle,
    val getters: List<Pair<String, MethodHandle>>
)

// Intercepts ALL network packets and extracts fields via MethodHandles (no hot-path reflection)
class PacketTelemetryListener(
    private val queue: ConcurrentLinkedQueue<TelemetryEvent>,
    private val serverId: String
) : PacketListener {

    private val lookup = MethodHandles.lookup()

    // Cache of pre-compiled extractors per wrapper class (built once, reused forever)
    private val extractorCache = ConcurrentHashMap<Class<*>, FieldExtractor?>()

    // Methods inherited from PacketWrapper internals that don't carry useful telemetry
    private val ignoredMethods = setOf(
        "getBuffer", "getServerVersion", "getClientVersion",
        "getPacketId", "getNativePacketId", "getPacketType",
        "getClass", "getChannel"
    )

    override fun onUserLogin(event: UserLoginEvent) {
        val user = event.getUser()
        val telemetryEvent = TelemetryEvent(
            type = "player_join",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = user.name
        )
        queue.add(telemetryEvent)
    }

    override fun onUserDisconnect(event: UserDisconnectEvent) {
        val user = event.getUser()
        val telemetryEvent = TelemetryEvent(
            type = "player_quit",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = user.name
        )
        queue.add(telemetryEvent)
    }

    override fun onPacketReceive(event: PacketReceiveEvent) {
        val packetName = event.packetType.name
        val userName = event.getUser().name
        // Extract fields using pre-compiled MethodHandles (no per-packet reflection)
        val fields = extractFields(event.packetType, event)
        val telemetryEvent = TelemetryEvent(
            type = "packet_in:$packetName",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = "$userName|$fields"
        )
        queue.add(telemetryEvent)
    }

    override fun onPacketSend(event: PacketSendEvent) {
        val packetName = event.packetType.name
        val userName = event.getUser().name
        // Extract fields using pre-compiled MethodHandles (no per-packet reflection)
        val fields = extractFields(event.packetType, event)
        val telemetryEvent = TelemetryEvent(
            type = "packet_out:$packetName",
            serverId = serverId,
            timestamp = System.currentTimeMillis(),
            payload = "$userName|$fields"
        )
        queue.add(telemetryEvent)
    }

    private fun extractFields(packetType: PacketTypeCommon, event: Any): String {
        val wrapperClass = packetType.wrapperClass ?: return "raw"
        // Get or build the extractor (reflection runs only on first encounter)
        val extractor = extractorCache.getOrPut(wrapperClass) { buildExtractor(wrapperClass, event) }
            ?: return "raw"
        return try {
            // Instantiate wrapper via pre-compiled constructor handle
            val wrapper = extractor.constructorHandle.invoke(event)
            // Call each getter via pre-compiled method handles
            val fields = extractor.getters.mapNotNull { (name, handle) ->
                try {
                    val value = handle.invoke(wrapper)
                    "$name=${valueToString(value)}"
                } catch (_: Throwable) {
                    null
                }
            }
            fields.joinToString(",")
        } catch (_: Throwable) {
            "raw"
        }
    }

    // Runs ONCE per packet type — uses reflection to discover, then compiles to MethodHandles
    private fun buildExtractor(wrapperClass: Class<*>, event: Any): FieldExtractor? {
        return try {
            // Find and compile the constructor into a MethodHandle
            val constructor = wrapperClass.getConstructor(event::class.java)
            val constructorHandle = lookup.unreflectConstructor(constructor)
            // Find getter methods, filter internals, compile each to a MethodHandle
            val getters = wrapperClass.methods
                .filter { it.name.startsWith("get") }
                .filter { it.parameterCount == 0 }
                .filter { it.name !in ignoredMethods }
                .filter { !PacketWrapper::class.java.methods.any { base -> base.name == it.name } || isOverridden(it, wrapperClass) }
                .sortedBy { it.name }
                .map { method ->
                    val key = method.name.removePrefix("get").replaceFirstChar { it.lowercase() }
                    // Compile the Method into a MethodHandle (JIT-inlinable after warmup)
                    val handle = lookup.unreflect(method)
                    Pair(key, handle)
                }
            FieldExtractor(constructorHandle, getters)
        } catch (_: Exception) {
            null
        }
    }

    private fun isOverridden(method: java.lang.reflect.Method, wrapperClass: Class<*>): Boolean {
        // Check if the method is declared in the wrapper subclass (not just inherited)
        return try {
            wrapperClass.getDeclaredMethod(method.name, *method.parameterTypes)
            true
        } catch (_: NoSuchMethodException) {
            false
        }
    }

    private fun valueToString(value: Any?): String {
        return when (value) {
            null -> "null"
            is ByteArray -> "[${value.size} bytes]"
            is Collection<*> -> "[${value.size} items]"
            is Array<*> -> "[${value.size} items]"
            else -> value.toString().take(128)
        }
    }
}

package com.example.telemetry

import io.lettuce.core.XAddArgs
import io.lettuce.core.api.StatefulRedisConnection
import kotlinx.serialization.ExperimentalSerializationApi
import kotlinx.serialization.protobuf.ProtoBuf
import org.bukkit.plugin.java.JavaPlugin
import java.util.Base64
import java.util.concurrent.ConcurrentLinkedQueue

// Drains the event queue on a background thread and publishes to Valkey Streams
class FlushLoop(
    private val plugin: JavaPlugin,
    private val queue: ConcurrentLinkedQueue<TelemetryEvent>,
    private val connection: StatefulRedisConnection<String, String>,
    private val serverId: String
) {

    @OptIn(ExperimentalSerializationApi::class)
    fun start() {
        // Schedule an async repeating task: 0 tick delay, 5 tick period (250ms at 20 TPS)
        plugin.server.scheduler.runTaskTimerAsynchronously(plugin, Runnable {
            val commands = connection.async()
            // Drain all queued events in one pass
            while (true) {
                val event = queue.poll() ?: break
                // Serialize the TelemetryEvent to Protobuf bytes
                val encoded = ProtoBuf.encodeToByteArray(TelemetryEvent.serializer(), event)
                // Base64-encode for safe storage in a String-codec connection
                val base64 = Base64.getEncoder().encodeToString(encoded)
                // Publish to the server-specific stream with bounded retention
                val args = XAddArgs.Builder.maxlen(10000).approximateTrimming()
                commands.xadd("telemetry:$serverId", args, mapOf("data" to base64))
            }
        }, 0L, 5L)
    }
}

package com.example.telemetry

import org.bukkit.Bukkit
import org.bukkit.plugin.java.JavaPlugin
import java.util.concurrent.ConcurrentLinkedQueue

// Samples server TPS and MSPT every 500ms and enqueues as telemetry events
class TpsSampler(
    private val plugin: JavaPlugin,
    private val queue: ConcurrentLinkedQueue<TelemetryEvent>,
    private val serverId: String
) {

    fun start() {
        // Schedule an async repeating task: 0 tick delay, 10 tick period (500ms at 20 TPS)
        plugin.server.scheduler.runTaskTimerAsynchronously(plugin, Runnable {
            // Read server performance metrics
            val tps = Bukkit.getTPS()[0]
            val mspt = Bukkit.getAverageTickTime()
            // Pack TPS and MSPT into the payload as a delimited string
            val event = TelemetryEvent(
                type = "server_tick",
                serverId = serverId,
                timestamp = System.currentTimeMillis(),
                payload = "$tps:$mspt"
            )
            queue.add(event)
        }, 0L, 10L)
    }
}

package com.example.telemetry

import com.github.retrooper.packetevents.PacketEvents
import com.github.retrooper.packetevents.event.PacketListenerPriority
import io.github.retrooper.packetevents.factory.spigot.SpigotPacketEventsBuilder
import io.lettuce.core.RedisClient
import io.lettuce.core.api.StatefulRedisConnection
import org.bukkit.plugin.java.JavaPlugin
import java.util.concurrent.ConcurrentLinkedQueue

// Main plugin class that initializes PacketEvents, Valkey, and all telemetry components
class TelemetryPlugin : JavaPlugin() {

    private lateinit var redisClient: RedisClient
    private lateinit var connection: StatefulRedisConnection<String, String>
    private val eventQueue = ConcurrentLinkedQueue<TelemetryEvent>()
    private val serverId = "server-1"

    override fun onLoad() {
        // Build and load the PacketEvents API instance
        PacketEvents.setAPI(SpigotPacketEventsBuilder.build(this))
        PacketEvents.getAPI().load()
    }

    override fun onEnable() {
        // Register the packet telemetry listener before initializing
        val listener = PacketTelemetryListener(eventQueue, serverId)
        PacketEvents.getAPI().eventManager.registerListener(listener, PacketListenerPriority.NORMAL)

        // Initialize PacketEvents (must happen after listener registration)
        PacketEvents.getAPI().init()

        // Initialize Lettuce client connecting to Valkey on localhost
        redisClient = RedisClient.create("redis://localhost:6379")
        connection = redisClient.connect()

        // Start the background flush loop (drains queue every 250ms)
        val flushLoop = FlushLoop(this, eventQueue, connection, serverId)
        flushLoop.start()

        // Start the TPS/MSPT sampler (samples every 500ms)
        val tpsSampler = TpsSampler(this, eventQueue, serverId)
        tpsSampler.start()

        logger.info("TelemetryPlugin enabled - streaming to Valkey via PacketEvents")
    }

    override fun onDisable() {
        // Terminate PacketEvents cleanly
        PacketEvents.getAPI().terminate()
        // Clean up Valkey connection on plugin disable
        connection.close()
        redisClient.shutdown()
        logger.info("TelemetryPlugin disabled - Valkey connection closed")
    }
}

Your telemetry producer is complete. It captures events without touching TPS, serializes them with Protobuf, and streams them into Valkey with bounded retention. Next up, you will build the Ktor analytics service that consumes these streams and pushes live data to the dashboard.

Build the Ktor Analytics Consumer

Your Paper plugin is now collecting events and publishing them to Valkey Streams. The next piece of the pipeline is the service that actually reads those events and makes sense of them.

In this step, you will build the Ktor analytics service that connects to Valkey, pulls event batches using consumer groups, aggregates the data in memory, and broadcasts live snapshots to connected dashboards over WebSocket.

In this step, get ready to:

• Set up a Valkey Streams consumer with consumer groups and batch reading.
• Build an in-memory aggregator that decodes Protobuf events and tracks metrics.
• Expose a WebSocket endpoint that pushes live data to dashboards every 500ms.

Set up the Valkey Streams consumer

The consumer connects to Valkey via Lettuce and uses consumer groups to coordinate reading. Consumer groups ensure that if you scale to multiple analytics service instances, each message is delivered to only one consumer.

• In your ktor-analytics-service module, create a new file StreamConsumer.kt inside src/main/kotlin/.
• Add the following code to set up the Redis connection and consumer loop:

package com.telemetry.analytics

import io.lettuce.core.Consumer
import io.lettuce.core.ExperimentalLettuceCoroutinesApi
import io.lettuce.core.RedisClient
import io.lettuce.core.XGroupCreateArgs
import io.lettuce.core.XReadArgs
import io.lettuce.core.XReadArgs.StreamOffset
import io.lettuce.core.api.coroutines
import io.lettuce.core.api.coroutines.RedisCoroutinesCommands
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.collect
import java.util.Base64

// Connects to Valkey and pulls telemetry events via consumer groups
@OptIn(ExperimentalLettuceCoroutinesApi::class)
object StreamConsumer {

    private val client = RedisClient.create("redis://localhost:6379")
    private val connection = client.connect()
    private val redis = connection.coroutines()

    // Create consumer group 'analytics' on each configured stream
    suspend fun initialize(streamKeys: List<String>) {
        for (key in streamKeys) {
            try {
                redis.xgroupCreate(
                    StreamOffset.from(key, "0-0"),
                    "analytics",
                    XGroupCreateArgs.Builder.mkstream()
                )
            } catch (e: Exception) {
                // Group already exists, safe to ignore
            }
        }
    }

    // Continuously read event batches from Valkey Streams
    fun startConsuming(scope: CoroutineScope, streamKeys: List<String>) {
        scope.launch(Dispatchers.IO) {
            // Build offsets to read only new messages for this group
            val offsets = streamKeys.map {
                StreamOffset.from(it, ">")
            }.toTypedArray()

            while (isActive) {
                // Pull up to 100 messages, blocking 200ms if none available
                val messages = redis.xreadgroup(
                    Consumer.from("analytics", "consumer1"),
                    XReadArgs.Builder.count(100).block(200),
                    *offsets
                )
                messages.collect { message ->
                    // Decode Base64 event data and pass to aggregator
                    val eventData = message.body["data"] ?: return@collect
                    val eventBytes = Base64.getDecoder().decode(eventData)
                    Aggregator.process(eventBytes)
                    // Acknowledge the processed message
                    redis.xack(message.stream, "analytics", message.id)
                }
            }
        }
    }

    fun shutdown() {
        connection.close()
        client.shutdown()
    }
}

What does this code do?

This object manages the entire Valkey consumption lifecycle for your analytics service.

• initialize() creates a consumer group called analytics on each stream key. The MKSTREAM option creates the stream if it does not exist yet.
• startConsuming() launches a coroutine on the IO dispatcher that loops indefinitely. Each iteration calls XREADGROUP with BLOCK 200ms and COUNT 100, pulling up to 100 messages or waiting 200ms if the stream is empty.
• StreamOffset.from(key, ">") tells Valkey to deliver only new, never-before-delivered messages to this consumer.
• After processing each message, xack acknowledges it so Valkey knows the message was handled and removes it from the pending entries list.

• Save StreamConsumer.kt.

Seeing unresolved reference errors?

Make sure your build.gradle.kts includes the io.lettuce:lettuce-core:7.6.0.RELEASE dependency from Step 1.

Also confirm you have kotlinx-coroutines-core and kotlinx-coroutines-reactive on your classpath. Lettuce's coroutines API requires both.

help me fix unresolved references in StreamConsumer.kt

Build the in-memory aggregator

The aggregator receives raw Protobuf-encoded bytes, decodes each TelemetryEvent envelope, and updates rolling counters. It tracks events per server, events per type, TPS readings, and player counts.

• Create a new file Aggregator.kt inside src/main/kotlin/ in your ktor-analytics-service module.
• Add the following code to decode events and maintain aggregation state:

package com.telemetry.analytics

import kotlinx.serialization.ExperimentalSerializationApi
import kotlinx.serialization.Serializable
import kotlinx.serialization.decodeFromByteArray
import kotlinx.serialization.protobuf.ProtoBuf
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong

// Tracks per-server and per-type event counts plus server health metrics
@OptIn(ExperimentalSerializationApi::class)
object Aggregator {

    // Per-server total event counts
    private val eventsPerServer = ConcurrentHashMap<String, AtomicLong>()
    // Per-type event counts (player_join, player_quit, entity_death, server_tick)
    private val eventsPerType = ConcurrentHashMap<String, AtomicLong>()
    // Latest TPS reading per server
    private val tpsPerServer = ConcurrentHashMap<String, Double>()
    // Current player count per server
    private val playerCountPerServer = ConcurrentHashMap<String, Int>()

    // Decode a Protobuf-encoded TelemetryEvent and update counters
    fun process(eventBytes: ByteArray) {
        val event = ProtoBuf.decodeFromByteArray<TelemetryEvent>(eventBytes)

        // Increment per-server counter
        eventsPerServer.getOrPut(event.serverId) { AtomicLong(0) }.incrementAndGet()
        // Increment per-type counter
        eventsPerType.getOrPut(event.type) { AtomicLong(0) }.incrementAndGet()

        // Update server health snapshots based on event type
        when (event.type) {
            "server_tick" -> {
                // Payload is already a String in "tps:mspt" format
                val parts = event.payload.split(":")
                if (parts.size >= 2) {
                    tpsPerServer[event.serverId] = parts[0].toDoubleOrNull() ?: 20.0
                }
            }
            "player_join" -> {
                playerCountPerServer[event.serverId] =
                    (playerCountPerServer[event.serverId] ?: 0) + 1
            }
            "player_quit" -> {
                playerCountPerServer[event.serverId] =
                    ((playerCountPerServer[event.serverId] ?: 0) - 1).coerceAtLeast(0)
            }
        }
    }

    // Return current state as a serializable snapshot
    fun getSnapshot(): AggregationSnapshot {
        return AggregationSnapshot(
            eventsPerServer = eventsPerServer.mapValues { it.value.get() },
            eventsPerType = eventsPerType.mapValues { it.value.get() },
            tpsPerServer = tpsPerServer.toMap(),
            playerCountPerServer = playerCountPerServer.toMap()
        )
    }
}

// JSON-serializable snapshot of the current aggregation state
@Serializable
data class AggregationSnapshot(
    val eventsPerServer: Map<String, Long>,
    val eventsPerType: Map<String, Long>,
    val tpsPerServer: Map<String, Double>,
    val playerCountPerServer: Map<String, Int>
)

What does this code do?

The Aggregator is the brain of your analytics service. It maintains four concurrent maps that build up a picture of your network's health in real time.

• ProtoBuf.decodeFromByteArray<TelemetryEvent>(eventBytes) deserializes the raw bytes back into the shared TelemetryEvent data class. This is the inverse of what the Paper plugin's flush loop does.
• The when block routes different event types to different counter updates. Server tick events update TPS. Join/quit events adjust the player count.
• getSnapshot() captures a point-in-time copy of all counters. This snapshot is what gets sent to the dashboard over WebSocket.
• ConcurrentHashMap and AtomicLong ensure thread safety since the consumer coroutine writes while the broadcast coroutine reads.

• Save Aggregator.kt.

Unresolved reference to TelemetryEvent?

The TelemetryEvent data class lives in your shared module. Make sure your ktor-analytics-service module's build.gradle.kts declares a dependency on the shared module (e.g. implementation(project(":shared"))).

help me fix unresolved TelemetryEvent reference in my Ktor analytics module

Expose a WebSocket endpoint at /ws/dashboard

The final piece ties everything together. The Ktor application starts the consumer, then broadcasts the aggregation snapshot to all connected dashboard clients every 500ms.

• Create Application.kt inside src/main/kotlin/ in your ktor-analytics-service module.
• Add the following code to wire up the server, consumer, and WebSocket broadcast:

package com.telemetry.analytics

import io.ktor.server.application.*
import io.ktor.server.engine.*
import io.ktor.server.netty.*
import io.ktor.server.routing.*
import io.ktor.server.websocket.*
import io.ktor.websocket.*
import kotlinx.coroutines.*
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
import java.util.Collections

// Entry point for the Ktor analytics service
fun main() {
    embeddedServer(Netty, port = 8080, module = Application::module)
        .start(wait = true)
}

fun Application.module() {
    // Install WebSocket support for real-time push
    install(WebSockets)

    // Thread-safe set of all connected dashboard sessions
    val sessions = Collections.synchronizedSet(mutableSetOf<WebSocketSession>())

    routing {
        // Dashboard clients connect here to receive live updates
        webSocket("/ws/dashboard") {
            sessions.add(this)
            try {
                // Keep the session open until client disconnects
                for (frame in incoming) { /* hold connection open */ }
            } finally {
                sessions.remove(this)
            }
        }
    }

    // Configure which streams to consume from
    val streamKeys = listOf("telemetry:lobby", "telemetry:survival")

    // Start the consumer and broadcast loop in the application scope
    launch {
        StreamConsumer.initialize(streamKeys)
        StreamConsumer.startConsuming(this, streamKeys)

        // Broadcast aggregation snapshot to all dashboards every 500ms
        while (isActive) {
            delay(500)
            val snapshot = Aggregator.getSnapshot()
            val json = Json.encodeToString(snapshot)
            sessions.forEach { session ->
                launch { session.send(json) }
            }
        }
    }
}

What does this code do?

This is the main entry point for your analytics service. It sets up three responsibilities in one place.

• install(WebSockets) enables the WebSocket plugin on the Ktor server. Without this, the webSocket() route would fail.
• The webSocket("/ws/dashboard") route adds each connecting client to a shared sessions set. The for (frame in incoming) loop keeps the connection alive until the client disconnects.
• The launch block starts the Redis consumer and then enters a broadcast loop. Every 500ms it serializes the current AggregationSnapshot to JSON and pushes it to every connected WebSocket session.
• Because Ktor's Application implements CoroutineScope, calling launch directly in the module function is safe. The coroutine is automatically cancelled when the server shuts down.

• Save Application.kt.
• Run the Ktor service from IntelliJ by clicking the green play button next to the main() function, or by running:

./gradlew :ktor-analytics-service:run

• You should see output indicating the server started on port 8080. The service is now consuming from Redis and ready to accept WebSocket connections.

Server fails to start on port 8080?

If you see an address-already-in-use error, another process is using port 8080. Stop it or change the port number in Application.kt.

If the service starts but immediately logs a Valkey connection error, confirm your valkey-telemetry container is still running by executing docker ps in your terminal.

help me troubleshoot my Ktor analytics service startup

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files and that your Ktor service is running on port 8080.

ⓧ I'd like to double check the full code

Here is the complete code for all three files in this step.

package com.telemetry.analytics

import io.lettuce.core.Consumer
import io.lettuce.core.ExperimentalLettuceCoroutinesApi
import io.lettuce.core.RedisClient
import io.lettuce.core.XGroupCreateArgs
import io.lettuce.core.XReadArgs
import io.lettuce.core.XReadArgs.StreamOffset
import io.lettuce.core.api.coroutines
import io.lettuce.core.api.coroutines.RedisCoroutinesCommands
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.collect
import java.util.Base64

// Connects to Valkey and pulls telemetry events via consumer groups
@OptIn(ExperimentalLettuceCoroutinesApi::class)
object StreamConsumer {

    private val client = RedisClient.create("redis://localhost:6379")
    private val connection = client.connect()
    private val redis = connection.coroutines()

    // Create consumer group 'analytics' on each configured stream
    suspend fun initialize(streamKeys: List<String>) {
        for (key in streamKeys) {
            try {
                redis.xgroupCreate(
                    StreamOffset.from(key, "0-0"),
                    "analytics",
                    XGroupCreateArgs.Builder.mkstream()
                )
            } catch (e: Exception) {
                // Group already exists, safe to ignore
            }
        }
    }

    // Continuously read event batches from Valkey Streams
    fun startConsuming(scope: CoroutineScope, streamKeys: List<String>) {
        scope.launch(Dispatchers.IO) {
            // Build offsets to read only new messages for this group
            val offsets = streamKeys.map {
                StreamOffset.from(it, ">")
            }.toTypedArray()

            while (isActive) {
                // Pull up to 100 messages, blocking 200ms if none available
                val messages = redis.xreadgroup(
                    Consumer.from("analytics", "consumer1"),
                    XReadArgs.Builder.count(100).block(200),
                    *offsets
                )
                messages.collect { message ->
                    // Decode Base64 event data and pass to aggregator
                    val eventData = message.body["data"] ?: return@collect
                    val eventBytes = Base64.getDecoder().decode(eventData)
                    Aggregator.process(eventBytes)
                    // Acknowledge the processed message
                    redis.xack(message.stream, "analytics", message.id)
                }
            }
        }
    }

    fun shutdown() {
        connection.close()
        client.shutdown()
    }
}

package com.telemetry.analytics

import kotlinx.serialization.ExperimentalSerializationApi
import kotlinx.serialization.Serializable
import kotlinx.serialization.decodeFromByteArray
import kotlinx.serialization.protobuf.ProtoBuf
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong

// Tracks per-server and per-type event counts plus server health metrics
@OptIn(ExperimentalSerializationApi::class)
object Aggregator {

    // Per-server total event counts
    private val eventsPerServer = ConcurrentHashMap<String, AtomicLong>()
    // Per-type event counts (player_join, player_quit, entity_death, server_tick)
    private val eventsPerType = ConcurrentHashMap<String, AtomicLong>()
    // Latest TPS reading per server
    private val tpsPerServer = ConcurrentHashMap<String, Double>()
    // Current player count per server
    private val playerCountPerServer = ConcurrentHashMap<String, Int>()

    // Decode a Protobuf-encoded TelemetryEvent and update counters
    fun process(eventBytes: ByteArray) {
        val event = ProtoBuf.decodeFromByteArray<TelemetryEvent>(eventBytes)

        // Increment per-server counter
        eventsPerServer.getOrPut(event.serverId) { AtomicLong(0) }.incrementAndGet()
        // Increment per-type counter
        eventsPerType.getOrPut(event.type) { AtomicLong(0) }.incrementAndGet()

        // Update server health snapshots based on event type
        when (event.type) {
            "server_tick" -> {
                // Payload is already a String in "tps:mspt" format
                val parts = event.payload.split(":")
                if (parts.size >= 2) {
                    tpsPerServer[event.serverId] = parts[0].toDoubleOrNull() ?: 20.0
                }
            }
            "player_join" -> {
                playerCountPerServer[event.serverId] =
                    (playerCountPerServer[event.serverId] ?: 0) + 1
            }
            "player_quit" -> {
                playerCountPerServer[event.serverId] =
                    ((playerCountPerServer[event.serverId] ?: 0) - 1).coerceAtLeast(0)
            }
        }
    }

    // Return current state as a serializable snapshot
    fun getSnapshot(): AggregationSnapshot {
        return AggregationSnapshot(
            eventsPerServer = eventsPerServer.mapValues { it.value.get() },
            eventsPerType = eventsPerType.mapValues { it.value.get() },
            tpsPerServer = tpsPerServer.toMap(),
            playerCountPerServer = playerCountPerServer.toMap()
        )
    }
}

// JSON-serializable snapshot of the current aggregation state
@Serializable
data class AggregationSnapshot(
    val eventsPerServer: Map<String, Long>,
    val eventsPerType: Map<String, Long>,
    val tpsPerServer: Map<String, Double>,
    val playerCountPerServer: Map<String, Int>
)

package com.telemetry.analytics

import io.ktor.server.application.*
import io.ktor.server.engine.*
import io.ktor.server.netty.*
import io.ktor.server.routing.*
import io.ktor.server.websocket.*
import io.ktor.websocket.*
import kotlinx.coroutines.*
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
import java.util.Collections

// Entry point for the Ktor analytics service
fun main() {
    embeddedServer(Netty, port = 8080, module = Application::module)
        .start(wait = true)
}

fun Application.module() {
    // Install WebSocket support for real-time push
    install(WebSockets)

    // Thread-safe set of all connected dashboard sessions
    val sessions = Collections.synchronizedSet(mutableSetOf<WebSocketSession>())

    routing {
        // Dashboard clients connect here to receive live updates
        webSocket("/ws/dashboard") {
            sessions.add(this)
            try {
                // Keep the session open until client disconnects
                for (frame in incoming) { /* hold connection open */ }
            } finally {
                sessions.remove(this)
            }
        }
    }

    // Configure which streams to consume from
    val streamKeys = listOf("telemetry:lobby", "telemetry:survival")

    // Start the consumer and broadcast loop in the application scope
    launch {
        StreamConsumer.initialize(streamKeys)
        StreamConsumer.startConsuming(this, streamKeys)

        // Broadcast aggregation snapshot to all dashboards every 500ms
        while (isActive) {
            delay(500)
            val snapshot = Aggregator.getSnapshot()
            val json = Json.encodeToString(snapshot)
            sessions.forEach { session ->
                launch { session.send(json) }
            }
        }
    }
}

Your analytics consumer is now live and reading from Valkey Streams. Next, you will build the real-time dashboard that connects over WebSocket to visualize all this data as it flows through the pipeline.

Build the Real-Time Dashboard

Your analytics service is live and broadcasting snapshots every 500ms over WebSocket.

Now you need somewhere for those snapshots to land. In this step, you will build a browser-based dashboard that connects to the Ktor WebSocket endpoint, parses the incoming JSON, and renders live charts that update at sub-second intervals.

In this step, get ready to:

• Create an HTML page with Chart.js showing player counts, TPS, and event throughput.
• Connect a WebSocket client that updates all charts every 500ms with smooth transitions.
• Test the full pipeline end-to-end from Minecraft events to live dashboard.

Create the HTML page with Chart.js

The dashboard is a single HTML file that loads Chart.js from a CDN and renders three visualizations: a line chart tracking player counts per server, a doughnut gauge for average TPS, and a numeric counter for total events per second.

• Create a dashboard folder alongside your existing project folders by running:

mkdir dashboard

• Create index.html inside the dashboard folder and add the following HTML structure:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Minecraft Telemetry Dashboard</title>
    <!-- Load Chart.js from jsDelivr CDN -->
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        body {
            font-family: system-ui, -apple-system, sans-serif;
            background: #1a1a2e;
            color: #e0e0e0;
            margin: 0;
            padding: 20px;
        }
        h1 {
            text-align: center;
            color: #00d4aa;
        }
        .grid {
            display: grid;
            grid-template-columns: 2fr 1fr;
            gap: 20px;
            max-width: 1200px;
            margin: 0 auto;
        }
        .card {
            background: #16213e;
            border-radius: 12px;
            padding: 20px;
        }
        .counter {
            text-align: center;
            font-size: 3rem;
            font-weight: bold;
            color: #00d4aa;
        }
        .counter-label {
            text-align: center;
            font-size: 1rem;
            color: #888;
        }
        canvas {
            max-height: 300px;
        }
    </style>
</head>
<body>
    <h1>Minecraft Telemetry Dashboard</h1>
    <div class="grid">
        <!-- Player count line chart -->
        <div class="card">
            <canvas id="playerChart"></canvas>
        </div>
        <!-- TPS gauge and events counter -->
        <div class="card">
            <canvas id="tpsChart"></canvas>
            <div class="counter" id="eventsCounter">0</div>
            <div class="counter-label">events/sec</div>
        </div>
    </div>

    <script>
    // Chart initialization and WebSocket logic goes here
    </script>
</body>
</html>

What does this HTML structure do?

The page loads Chart.js from the official jsDelivr CDN so you do not need any build tools or npm installs.

The two-column grid gives the player count line chart most of the space (left), while the TPS gauge and events counter sit in a narrower card on the right. Each chart renders into a <canvas> element that Chart.js controls.

• Now replace the placeholder // Chart initialization and WebSocket logic goes here comment inside the <script> tag with the following Chart.js initialization code:

    // Line chart for player counts per server
    const playerCtx = document.getElementById('playerChart').getContext('2d');
    const playerChart = new Chart(playerCtx, {
        type: 'line',
        data: {
            labels: [],
            datasets: []
        },
        options: {
            responsive: true,
            animation: { duration: 400 },
            plugins: {
                title: { display: true, text: 'Player Count Per Server', color: '#e0e0e0' }
            },
            scales: {
                x: { ticks: { color: '#888' }, grid: { color: '#333' } },
                y: { beginAtZero: true, ticks: { color: '#888' }, grid: { color: '#333' } }
            }
        }
    });

    // Doughnut chart as TPS gauge (half-circle)
    const tpsCtx = document.getElementById('tpsChart').getContext('2d');
    const tpsChart = new Chart(tpsCtx, {
        type: 'doughnut',
        data: {
            labels: ['TPS', 'Remaining'],
            datasets: [{
                data: [20, 0],
                backgroundColor: ['#00d4aa', '#333'],
                borderWidth: 0
            }]
        },
        options: {
            responsive: true,
            animation: { duration: 400 },
            circumference: 180,
            rotation: 270,
            plugins: {
                title: { display: true, text: 'Average TPS', color: '#e0e0e0' }
            }
        }
    });

    // Sliding window and color assignment for server datasets
    const maxDataPoints = 60;
    const serverColors = {};
    const colorPalette = ['#00d4aa', '#ff6384', '#36a2eb', '#ffce56', '#9966ff'];
    let colorIndex = 0;

    // Get or assign a consistent color for each server
    function getServerColor(serverId) {
        if (!serverColors[serverId]) {
            serverColors[serverId] = colorPalette[colorIndex % colorPalette.length];
            colorIndex++;
        }
        return serverColors[serverId];
    }

What does this code do?

The playerChart is a line chart with empty datasets. New datasets get added dynamically as servers come online. The 400ms animation duration gives smooth transitions without lag.

The tpsChart uses a half-doughnut (180 degrees) to act as a gauge from 0 to 20 TPS. The getServerColor function assigns each server a consistent color so datasets stay visually stable across updates.

• Save dashboard/index.html.
• Open the file in your browser. You should see a dark page with the title "Minecraft Telemetry Dashboard", an empty line chart labeled "Player Count Per Server" on the left, and a half-circle TPS gauge showing 20/20 on the right with a "0 events/sec" counter below it.

Charts not rendering?

Make sure you opened the actual index.html file in your browser (check the URL bar shows a file path). If you see a blank page, open the browser console and check for script loading errors. The CDN requires an internet connection.

help me debug why my Chart.js charts are not rendering

Connect the WebSocket client

The charts are ready but static. The WebSocket client connects to your Ktor service and pushes every incoming snapshot into the charts, making the dashboard update live every 500ms.

• In dashboard/index.html, add the following WebSocket connection code directly below the getServerColor function (still inside the <script> tag):

    // Connect to the Ktor WebSocket endpoint
    const ws = new WebSocket('ws://localhost:8080/ws/dashboard');

    ws.onmessage = (event) => {
        const snapshot = JSON.parse(event.data);

        // Update player count line chart
        const now = new Date().toLocaleTimeString();
        playerChart.data.labels.push(now);
        if (playerChart.data.labels.length > maxDataPoints) {
            playerChart.data.labels.shift();
        }

        // Add or update datasets for each server
        for (const [serverId, count] of Object.entries(snapshot.playerCountPerServer)) {
            let dataset = playerChart.data.datasets.find(ds => ds.label === serverId);
            if (!dataset) {
                dataset = {
                    label: serverId,
                    data: [],
                    borderColor: getServerColor(serverId),
                    backgroundColor: getServerColor(serverId) + '33',
                    tension: 0.3,
                    fill: true
                };
                playerChart.data.datasets.push(dataset);
            }
            dataset.data.push(count);
            if (dataset.data.length > maxDataPoints) {
                dataset.data.shift();
            }
        }
        playerChart.update();

        // Update TPS gauge with average across all servers
        const tpsValues = Object.values(snapshot.tpsPerServer);
        const avgTps = tpsValues.length > 0
            ? tpsValues.reduce((a, b) => a + b, 0) / tpsValues.length
            : 0;
        tpsChart.data.datasets[0].data = [avgTps, 20 - avgTps];
        tpsChart.update();

        // Update events/second counter
        const totalEvents = Object.values(snapshot.eventsPerServer)
            .reduce((a, b) => a + b, 0);
        document.getElementById('eventsCounter').textContent = totalEvents;
    };

    // Auto-reconnect if WebSocket drops
    ws.onclose = () => {
        console.log('WebSocket disconnected. Reconnecting in 2s...');
        setTimeout(() => location.reload(), 2000);
    };

What does this code do?

On every WebSocket message (arriving every 500ms from Ktor), the handler parses the JSON snapshot and pushes data into the charts. The maxDataPoints sliding window of 60 keeps the line chart scrolling smoothly without growing unbounded.

For the TPS gauge, it averages TPS across all connected servers and fills the doughnut proportionally out of 20. The onclose handler reloads the page after 2 seconds if the connection drops, providing basic auto-reconnect.

• Save dashboard/index.html.
• If your Ktor service is running, open the browser console. You should see no errors, and the WebSocket connection status should show as open. Charts will remain empty until telemetry events flow through Redis.

WebSocket connection failing?

Confirm your Ktor analytics service is running on port 8080. If you see a "connection refused" error in the console, start the Ktor service first. The dashboard will auto-reload once the service comes up.

help me debug my WebSocket connection to the Ktor analytics service

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboard/index.html with all the chart initialization and WebSocket code in place.

ⓧ I'd like to double check the full code

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Minecraft Telemetry Dashboard</title>
    <!-- Load Chart.js from jsDelivr CDN -->
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        body {
            font-family: system-ui, -apple-system, sans-serif;
            background: #1a1a2e;
            color: #e0e0e0;
            margin: 0;
            padding: 20px;
        }
        h1 {
            text-align: center;
            color: #00d4aa;
        }
        .grid {
            display: grid;
            grid-template-columns: 2fr 1fr;
            gap: 20px;
            max-width: 1200px;
            margin: 0 auto;
        }
        .card {
            background: #16213e;
            border-radius: 12px;
            padding: 20px;
        }
        .counter {
            text-align: center;
            font-size: 3rem;
            font-weight: bold;
            color: #00d4aa;
        }
        .counter-label {
            text-align: center;
            font-size: 1rem;
            color: #888;
        }
        canvas {
            max-height: 300px;
        }
    </style>
</head>
<body>
    <h1>Minecraft Telemetry Dashboard</h1>
    <div class="grid">
        <!-- Player count line chart -->
        <div class="card">
            <canvas id="playerChart"></canvas>
        </div>
        <!-- TPS gauge and events counter -->
        <div class="card">
            <canvas id="tpsChart"></canvas>
            <div class="counter" id="eventsCounter">0</div>
            <div class="counter-label">events/sec</div>
        </div>
    </div>

    <script>
    // Line chart for player counts per server
    const playerCtx = document.getElementById('playerChart').getContext('2d');
    const playerChart = new Chart(playerCtx, {
        type: 'line',
        data: {
            labels: [],
            datasets: []
        },
        options: {
            responsive: true,
            animation: { duration: 400 },
            plugins: {
                title: { display: true, text: 'Player Count Per Server', color: '#e0e0e0' }
            },
            scales: {
                x: { ticks: { color: '#888' }, grid: { color: '#333' } },
                y: { beginAtZero: true, ticks: { color: '#888' }, grid: { color: '#333' } }
            }
        }
    });

    // Doughnut chart as TPS gauge (half-circle)
    const tpsCtx = document.getElementById('tpsChart').getContext('2d');
    const tpsChart = new Chart(tpsCtx, {
        type: 'doughnut',
        data: {
            labels: ['TPS', 'Remaining'],
            datasets: [{
                data: [20, 0],
                backgroundColor: ['#00d4aa', '#333'],
                borderWidth: 0
            }]
        },
        options: {
            responsive: true,
            animation: { duration: 400 },
            circumference: 180,
            rotation: 270,
            plugins: {
                title: { display: true, text: 'Average TPS', color: '#e0e0e0' }
            }
        }
    });

    // Sliding window and color assignment for server datasets
    const maxDataPoints = 60;
    const serverColors = {};
    const colorPalette = ['#00d4aa', '#ff6384', '#36a2eb', '#ffce56', '#9966ff'];
    let colorIndex = 0;

    // Get or assign a consistent color for each server
    function getServerColor(serverId) {
        if (!serverColors[serverId]) {
            serverColors[serverId] = colorPalette[colorIndex % colorPalette.length];
            colorIndex++;
        }
        return serverColors[serverId];
    }

    // Connect to the Ktor WebSocket endpoint
    const ws = new WebSocket('ws://localhost:8080/ws/dashboard');

    ws.onmessage = (event) => {
        const snapshot = JSON.parse(event.data);

        // Update player count line chart
        const now = new Date().toLocaleTimeString();
        playerChart.data.labels.push(now);
        if (playerChart.data.labels.length > maxDataPoints) {
            playerChart.data.labels.shift();
        }

        // Add or update datasets for each server
        for (const [serverId, count] of Object.entries(snapshot.playerCountPerServer)) {
            let dataset = playerChart.data.datasets.find(ds => ds.label === serverId);
            if (!dataset) {
                dataset = {
                    label: serverId,
                    data: [],
                    borderColor: getServerColor(serverId),
                    backgroundColor: getServerColor(serverId) + '33',
                    tension: 0.3,
                    fill: true
                };
                playerChart.data.datasets.push(dataset);
            }
            dataset.data.push(count);
            if (dataset.data.length > maxDataPoints) {
                dataset.data.shift();
            }
        }
        playerChart.update();

        // Update TPS gauge with average across all servers
        const tpsValues = Object.values(snapshot.tpsPerServer);
        const avgTps = tpsValues.length > 0
            ? tpsValues.reduce((a, b) => a + b, 0) / tpsValues.length
            : 0;
        tpsChart.data.datasets[0].data = [avgTps, 20 - avgTps];
        tpsChart.update();

        // Update events/second counter
        const totalEvents = Object.values(snapshot.eventsPerServer)
            .reduce((a, b) => a + b, 0);
        document.getElementById('eventsCounter').textContent = totalEvents;
    };

    // Auto-reconnect if WebSocket drops
    ws.onclose = () => {
        console.log('WebSocket disconnected. Reconnecting in 2s...');
        setTimeout(() => location.reload(), 2000);
    };
    </script>
</body>
</html>

Test the full pipeline

Every component of the pipeline is now built. Time to light it up end-to-end and watch events flow from a Minecraft server into your dashboard in real-time.

• Confirm Redis is running by executing:

docker exec redis-telemetry redis-cli PING

• You should see PONG in the output.
• Start the Ktor analytics service from IntelliJ IDEA (run Application.kt).
• Start your Paper server with the telemetry plugin deployed.
• Open dashboard/index.html in your browser.
• Join the Minecraft server, walk around, break blocks, and kill a mob.

Within half a second of joining, you should see the player count line chart add a new dataset for your server and start climbing. The TPS gauge should hold steady near 20. The events/second counter should increment with each action you take in-game.

What should I see?

• The line chart adds a colored line labeled with your server ID. Each 500ms tick adds a new data point.
• The TPS gauge stays green near 20 (the plugin runs entirely off the main thread, so TPS is unaffected).
• The events counter climbs steadily. Killing mobs and joining/leaving produces visible spikes.
• When you disconnect from the Minecraft server, the player count line drops to 0 on the next tick.

🙋‍♀️ Dashboard staying blank after joining?

Check the Ktor service terminal for errors. If XREADGROUP is failing, the consumer group may not have been created. Restart the Ktor service so it re-runs the XGROUP CREATE on startup.

Also verify the Paper plugin loaded correctly by checking the server console for your plugin's enable message. If the plugin is not publishing events, the Ktor consumer has nothing to forward.

help me debug why my telemetry dashboard is not receiving data from the pipeline

Secret mission

Per-Server Breakdown and Historical Sparklines

Your dashboard shows the live pulse of the network, but operations teams need to spot trends over time. Can you add a sliding-window history buffer, per-server event breakdowns, and sparkline mini-charts that reveal the last 60 seconds of activity at a glance?

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally with a self-hosted Valkey container, so there are no ongoing cloud costs.

Resources you used:

• Valkey Docker container (valkey-telemetry) running on port 6379.
• Local project files: ktor-analytics-service/, paper-plugin/, dashboard/, and shared/ directories.

Keep everything running

No action needed. Choose this if you're still actively building or testing your telemetry pipeline.

• Valkey uses minimal resources (approximately 50MB RAM). It's fine to leave the container running for development.
• Your project files remain on disk ready for the next coding session.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping everything intact for later.

• Stop the Valkey container by running:

docker stop valkey-telemetry

• When you're ready to resume, start it again by running:

docker start valkey-telemetry

• Your stream data persists inside the container. You can pick up exactly where you left off.

Delete - I don't want to use this again

Remove all project resources and start fresh.

• Remove the Valkey container entirely by running:

docker rm -f valkey-telemetry

• Delete the project directories from your Desktop by running:

cd ~/Desktop
rm -rf ktor-analytics-service paper-plugin dashboard shared

Windows users

Use this command instead to remove the project directories:

cd ~\Desktop
rmdir /s /q ktor-analytics-service paper-plugin dashboard shared

That's a Wrap!

That's a Wrap!

Nice work. You just built a production-grade real-time telemetry pipeline that streams thousands of events per second from Minecraft servers into a live analytics dashboard with sub-second latency.

You've learned how to:

• Build an async Paper plugin that collects player events and server metrics completely off the main thread using a ConcurrentLinkedQueue and background flush loop, achieving zero TPS impact.
• Design a Protobuf envelope pattern that publishes high-throughput telemetry to Valkey Streams with bounded retention and consumer group delivery guarantees.
• Build a Ktor analytics service that consumes streams, aggregates metrics in memory, and pushes live snapshots to a Chart.js dashboard over WebSocket every 500ms.
• Secret Mission: Added per-server event breakdown charts with historical sparklines for trend visualization.

Ready to quiz yourself?