DOCSEARCHbyAlgolia
Submit Docs

KP Library

website·Indexed 30 days ago

https://honestbank.github.io/kp

A library for processing messages from Kafka topics with an intuitive API. It features a robust middleware system for implementing backoff policies, retries, dead-lettering, graceful shutdown, tracing, and metrics. The library includes a Consumer interface for retrieving and committing messages and a Producer capable of producing Avro formatted messages via a schema registry.

Tokens
8.6K
Snippets
28
Records
46
Agent score
74%

What's inside KP

  1. Understand Kafka Replication and High Availability

    Kafka replicates messages across multiple brokers to ensure high availability and recovery in the event of a broker failure. This replication architecture also supports the use of multiple independent consumer groups reading from the same topic for parallel processing.

  2. Understand Schema Registry requirements for KP

    KP requires a schema registry to ensure that only backwards compatible applications are deployed. The schema registry allows Kafka clients to publish message schemas for topics and manage schema evolution (e.g., maintaining backwards compatibility).

  3. Understand the KP middleware execution order

    The KP middleware system uses a stack-based execution model. Middleware functions are executed in the order they are added via the AddMiddleware method.

    Control flow follows a recursive pattern: each middleware executes its 'before' logic, calls the next function to trigger the subsequent middleware, and then executes its 'after' logic as the stack unwinds. This means the 'before' phase happens in the order of addition, while the 'after' phase happens in reverse order.

    Execution Sequence Example: If Middleware A, B, and C are added in that order, the flow is:

    1. Middleware A (before)
    2. Middleware B (before)
    3. Middleware C (before)
    4. Middleware C (after)
    5. Middleware B (after)
    6. Middleware A (after)
    package main

import "context"

type Middleware[T any] struct {
	Name string
}

func (m Middleware[T]) Process(ctx context.Context, item T, next func(ctx context.Context, item T) error) error {
	println("Middleware " + m.Name + " -> before logic")
	err := next(ctx, item)
	println("Middleware " + m.Name + " -> after logic")

	return err
}

func main() {
   kp := v2.New[string]()
   middlewareA := Middleware[string]{Name: "A"}
   middlewareB := Middleware[string]{Name: "B"}
   middlewareC := Middleware[string]{Name: "C"}
   
   // Order of addition determines execution order
   kp.AddMiddleware(middlewareA)
   kp.AddMiddleware(middlewareB)
   kp.AddMiddleware(middlewareC)
}

  4. Implement end-to-end distributed tracing with kp

    The kp library produces spans but requires the user to configure the exporter (where spans are sent and in what format). In a typical setup, Jaeger traces can be configured to be sent to Tempo. A complete end-to-end flow involves:

    1. Setting up a tracing exporter.
    2. Exposing trace-ids (e.g., via traceparent response headers) to the user.
    3. Producing messages into Kafka using a request context to maintain the trace.
    4. Configuring workers with retries and dead-letter queues while maintaining tracing context.

  5. Prerequisites for using KP

    To use the KP library, ensure your environment meets the following requirements:

    • Go version 1.18 or higher (required for generics support).
    • A Schema Registry to register topic schemas (currently mandatory).
    • cgo must be enabled during the build process of your Go code.

  6. Produce Avro-formatted messages with KP

    KP provides a producer capable of sending Avro-formatted messages to Kafka. This integration ensures backwards compatibility by utilizing a schema registry. A key feature is that initializing the producer automatically publishes the schema to the registry; if the schema contains a breaking change, the application startup will fail, preventing incompatible data from being produced.
    package main

import (
	"context"
	"github.com/honestbank/kp/v2/producer"
)

type UserLoggedIn struct {
	UserID string `avro:"user_id"`
}

func main() {
	p, err := producer.New[UserLoggedIn]("topic-name", getConfig())
	defer p.Flush()
	if err != nil {
		panic(err)
	}
	err = p.Produce(context.Background(), UserLoggedIn{})
	if err != nil {
		panic(err)
	}
}

func getConfig() any {
	return nil // return your config
}

  7. Implement a Dead Letter Queue using DeadletterMiddleware

    The DeadletterMiddleware sends messages to a designated deadletter topic after a specified number of processing retries have failed. This is used to isolate problematic messages for debugging and to prevent them from blocking processing pipelines.

    Key Requirements:

    • Producer Interface: A producer must be provided to handle the actual sending of messages to the deadletter topic.
    • Retry Threshold: A numeric value defining how many retries occur before the message is moved to the deadletter topic.
    • Error Handler (Optional): A function to handle errors that occur specifically when attempting to produce a message to the deadletter topic.

    Critical Integration Note: The DeadletterMiddleware must be placed after the retry middleware in the middleware chain. This ensures the retry logic is exhausted before the message is routed to the deadletter topic.

    Maintenance Warning: Messages sent to the deadletter topic are not automatically deleted from Kafka; a separate cleanup process must be implemented to prevent storage accumulation.

    // Initialize the deadletter middleware
// Parameters: producer, retry threshold, error handler function
deadletterMiddleware := middlewares.NewDeadletterMiddleware(myProducer, 5, myOnProduceErrorsFunc)

// Add the middleware to the processor
processor := kp.AddMiddleware(deadletterMiddleware)

  8. Configure OpenTelemetry TracerProvider for KP tracing

    To enable tracing, you need to set up a TracerProvider that exports spans to a backend (such as Jaeger). This setup must be performed before calling otel.GetTracerProvider() in the KP middleware configuration.
    func setupTracing() func() {
	exporter, err := jaeger.New(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint(os.Getenv("COLLECTOR_URL"))))
	if err != nil {
		panic(err)
	}

	resc := resource.Default()
	tp := trace.NewTracerProvider(
		race.WithBatcher(exporter),
		race.WithResource(resc),
		race.WithResource(resource.NewWithAttributes(semconv.SchemaURL, semconv.ServiceNameKey.String("form-service-web"))),
	)

	otel.SetTracerProvider(tp)

	return func() {
		exporter.Shutdown(context.Background())
		tp.Shutdown(context.Background())
	}
}

  9. Implement a custom middleware using KPMiddleware

    Middlewares in the KP library allow you to intercept and modify the processing of messages. To create a custom middleware, implement the KPMiddleware[T] interface. The Process method is executed for every message and provides a context for cancellation or value passing, the message item itself, and a next function to continue the middleware chain.
    type KPMiddleware[T any] interface {
	Process(ctx context.Context, item T, next func(ctx context.Context, item T) error) error
}

  10. Implement Kafka processing with KP middlewares

    KP is a framework for Kafka that separates business logic from infrastructure concerns like retries, backoffs, and tracing. By using a middleware-based approach, you can configure complex message-handling behaviors (such as exponential backoff or dead-lettering) in a single line of code, keeping your core processing function focused solely on the business logic.
    package main

import (
	"context"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/honestbank/kp/v2/middlewares/consumer"
	"github.com/honestbank/kp/v2/middlewares/deadletter"
	"github.com/honestbank/kp/v2/middlewares/retry"
	"time"

	backoff_policy "github.com/honestbank/backoff-policy"
	v2 "github.com/honestbank/kp/v2"
	"github.com/honestbank/kp/v2/middlewares/backoff"
	"github.com/honestbank/kp/v2/middlewares/measurement"
	"github.com/honestbank/kp/v2/middlewares/retry_count"
	"github.com/honestbank/kp/v2/middlewares/tracing"
	"go.opentelemetry.io/otel"
)

func main() {
	applicationName := "send-login-notification-worker"
	kp := v2.New[kafka.Message]()

	kp.
		AddMiddleware(consumer.NewConsumerMiddleware(getConsumer())).
		AddMiddleware(deadletter.NewDeadletterMiddleware(getDeadletterProducer(), 10, func(err error) {})).
		AddMiddleware(retry.NewRetryMiddleware(getRetryMiddleware(), func(err error) {})).
		AddMiddleware(backoff.NewBackoffMiddleware(backoff_policy.NewExponentialBackoffPolicy(time.Millisecond*200, 10))).
		AddMiddleware(tracing.NewTracingMiddleware(otel.GetTracerProvider())).
		AddMiddleware(measurement.NewMeasurementMiddleware("path/to/prometheus-push-gateway", applicationName)).
		AddMiddleware(retry_count.NewRetryCountMiddleware())

	kp.Process(func(ctx context.Context, message *kafka.Message) error {
		// Business logic goes here. 
		// Access current retry count via context:
		count := retry_count.FromContext(ctx)
		return nil
	})
}

  11. Best practices for KP migration deployment

    To minimize impact on end-users and prevent data processing errors during migration:

    1. Test the migration in a staging environment before deploying to production.
    2. Communicate the migration plan to team members and stakeholders.
    3. Schedule the migration during a maintenance window.

  12. Implement common middleware patterns in KP

    Depending on the desired behavior, KP middleware can be implemented using different logic placements within the Process method:

    • Before Logic Only: Used for preparation or injection. Examples: Consumer middleware (retrieving items), RetryCount middleware (injecting headers into context).
    • After Logic Only: Used for evaluation of results. Examples: Deadletter middleware (determining if a message should be dead-lettered), Retry middleware.
    • Both Before and After: Used for wrapping operations. Examples: Measurement (start timer before, measure duration after), Tracing (start span before, end span after), Backoff (delay execution and adjust intervals).
    • Flow Interruption: Middleware like Gracefulshutdown can stop the execution flow entirely by not calling the next function.