DEV Community: Emre Odabas

Kafka Exception C{r}onsumer 🔥 🚀

Emre Odabas — Fri, 11 Nov 2022 12:29:31 +0000

As a Trendyol Indexing Team, our architecture is heavily based on an event-based system that should be consistent, eventually. So even though this word seems to give relief, it is tough to ensure that every message in the system is successfully processed.

Our eventing system relies on Apache Kafka. It gives us powerful abilities when we need performance on event streaming. Our messages carry invalidations (daily 150M+) that should apply to Trendyol contents (300M+). This invalidation could be about a stock, promotion, or one of our 30 events.

In events worlds, we need to orchestrate our messages even in some inconsistent states. Retrying and delaying messages are our key strategies in this “eventual” life cycle of events. We have 10+ consumers that should apply this strategy. Therefore, we created an open-sourced library called Kafka Cronsumer for easy implementation.

How Kafka Cronsumer Works 💡

As the library name suggests, it consumes events within the exception topic based on the given cron expression.

For example, we could specify a cron expression as */20 * * * *, which means to run every 20th minutes. We could also set a duration 15 minutes (15m) represents our exception consumer actively consuming events within this fixed duration.

If the consumer gets an error when consuming a message, it increases the message retry count and produces the exception topic for handling the next iteration (next work time) again.
A message is discarded or moved to a dead letter if it exceeds maxRetry value.
Each message is processed only once at every iteration. So if the consumer encounters a message produced after our consumer start time, it pauses and waits for the next iteration.

As an overview, it works shown below.

How to use it? 👈

First of all, we need to set the required config values. We can also specify Kafka Consumer & Producer related configs, but it is not mandatory.

consumer:
  groupId: "exception-consumer"
  topic: "exception"
  maxRetry: 3
  concurrency: 1
  cron: "*/20 * * * *"
  duration: 15m

We define consumeFn function, which describes how the exception messages are consumed.

After that, we initialize Kafka cronsumer with these config values and consumeFn and run. It starts to work as shown above in the how to Kafka Cronsumer works section.

func main() {
        // ..
    var consumeFn kafka.ConsumeFn = func(message kafka.Message) error {
        fmt.Printf("consumer > Message received: %s\n", string(message.Value))
        return nil
    }

    c := cronsumer.New(kafkaConfig, consumeFn)
    c.Run()
}

You can find a number of ready-to-run examples at this directory.

When to use it? 🤔

Iteration-based back-off strategies are applicable
Messages could be processed in an eventually consistent state
Max retry exceeded messages could be ignored
To increase consumer resiliency -To increase consumer performance with concurrency

When to avoid?❗️

Messages should be processed in order
Messages should be certainly processed (we discard messages if max retry is exceeded)
Messages should be committed (we use auto-commit interval for increasing performance)
Messages with TTL (Time to Live)

Thanks for reading this far. All feedbacks are welcome. We love to share our knowledge. If you like to “share” with us, you could apply for our open positions.

Co-Authored by:
@abdulsametileri

Quick Guide: Go 1.19 Features

Emre Odabas — Mon, 18 Jul 2022 14:55:15 +0000

After the big bang of Go 1.18, Go still continues to develop new features. If you miss what comes with 1.18, check my previous Quick Guide. For now, RC1 has already been released and the version will be released in August 2022. Let’s look at what comes with Go 1.19.

Atomic Types

Bool, Int32,Int64, Uint32, Uint64, Uintptr and Pointer are new atomic types under sync/atomic package. If you are unfamiliar, we could summarize the atomic package that

provides low-level atomic memory primitives useful for implementing synchronization algorithms*.

With these new types, we could use Load,Store, Swap, CompareAndSwap functions. TheAdd function is also could be used for numeric types. These functions simplify the code we have written before Go 1.19, and you can check this new code on Go Playground.

Soft Memory Limit

As you may know, when it comes to tuning memory on Go, you have only one option to set GOGC(runtime/debug.SetGCPercent). Even though this optimization basically gives a tradeoff between CPU and memory, it does not respect users’ memory limit**.

As described in the proposal, we could define a soft memory limit with the GOMEMLIMIT environment variable or runtime/debug.SetMemoryLimitfunction.

With this new feature, we could decrease memory limits and define better memory utilization. It could be time to leave behind the memory ballast workaround the twitch team gave us🙏.

New Functions

Go 1.19 has several changes in core libraries. When I checked those, I select some of them seems significant to me.

fmt has new functions for appending data. Append, Appendf and Appendln
sort.Find is a new function for finding the smallest index of i with given comparator function. It offers better experience than sort.Search
Sorting algorithms switched to pattern-defeating quicksort for faster results for several scenarios.
New JoinPath function (URL.JoinPath method) creates a new URL by joining a list of path elements.

Unix build constraint

unix is a new build constraint that is satisfied if GOOS is one of

ios, linux, android, darwin, dragonfly, freebsd, hurd, illumos, netbsd, aix, openbsd or solaris

Compiler

The compiler now uses a jump table to implement large integer and string switch statements. Performance improvements for the switch statement vary but can be on the order of 20% faster. (GOARCH=amd64 and GOARCH=arm64 only)***.

You could check this benchmark for details.

With 1.19 version, the community works well and takes care of several performance optimizations. Many improvements and fixes will also be in this version. Those features are my selections. If I missed any fancy features, please contact me.

If you like to join our Gophers, you could apply for our open positions.

Quick Guide: Go 1.18 Features

Emre Odabas — Fri, 10 Dec 2021 20:30:17 +0000

With the Go 1.18 version, which is planned to be released in the first quarter of 2022, many new features are waiting for us.
This version will be one of the most critical packages of Go history and we will reach the beta version probably in a month. So what are these changes?

Type Parameters (Generics)

The long-awaited "Type Parameter" feature in Go is actually known as Generics in some other languages. With this feature, we can give dynamic types to our functions. Let's dive into a simple example and write a function that finds the minimum value of any two int variables.

func minInt(x int, y int) int {
   if x < y {
      return x
   } else {
      return y
   }
}

When we need the same functionality for float32, we need to rewrite the same thing also for float32.

func minFloat(x float32, y float32) float32 {
   if x < y {
      return x
   } else {
      return y
   }
}

With the power of Type Parameters, we could easily define a function that could get desired types of variables. For defining types, we also encounter a new operator called tilde ~. This operator actually returns us the interface of the type information, so we can make our type restrictions. Let's write the same min function that covers int and float32 types as below.

type OrderTypes interface { 
   ~int | ~float32 
}
func min[P OrderTypes](x, y P) P {
    if x < y {
      return x
    } else {
      return y
    }
}

It seems pretty solid, right? Go is also proposed a constraints library that contains already defined type groups. For example, we could implement our function for all ordered types.

func min[P constraints.Ordered](x, y P) P {
    if x < y {
      return x
    } else {
      return y
    }
}

Of course, the use of generics is not limited to this sample. It could be very useful especially in ORM-style problems that we generally create functions that work with multiple types. As well, you could check these examples shared by the Go team also give you some ideas. If you desire to watch the details, there is a great presentation the team made at GopherCon.
Apart from these, the online editor (playground) where we can experiment will also be useful. If you wish to try 1.18 in your own environment, you could download gotip and work with the active branch of Go.

Fuzzing Test

With the Golang 1.18 package, the "Fuzzing" feature will come into our lives. With this test library, which will be offered as beta, we will be able to automatically make random mutations of our inputs in unit tests.
As software developers, we sometimes miss unit tests edge cases. And this lack of data diversity could be abused and used in security leaks and vulnerabilities especially on critical libraries. We could prevent this situation using Fuzzing Test.
For implementing Fuzzing Test, you could wrap your unit tests with a method that name starts with the Fuzz prefix. You could also use the Go Testing page as a sample code.

func FuzzHex(f *testing.F) {
  for _, seed := range [][]byte{ {}, {0}, {9}, {0xa}, {0xf}, {1, 2, 3, 4}} {
    f.Add(seed)
  }
  f.Fuzz(func(t *testing.T, in []byte) {
    enc := hex.EncodeToString(in)
    out, err := hex.DecodeString(enc)
    if err != nil {
      t.Fatalf("%v: decode: %v", in, err)
    }
    if !bytes.Equal(in, out) {
      t.Fatalf("%v: not equal after round trip: %v", in, out)
    }
  })
}

Workspaces

This feature comes to facilitates developers' jobs, who work on multiple modules at the same time. Although the module version can be changed with the replace command in go.mod version 1.17, it is often forgotten to clear these replace commands while committing the code. Inserting or removing that line every time is a very bothering thing to do for these developers.
With the Workspace feature, you could create a new file called go.work and write the replace command there. This file allows us to change modules specific to our working environment without touching the existing go.mod file. As an example, you could look at the property's proposed design page.

go 1.17

directory (
    ./baz // foo.org/bar/baz
    ./tools // golang.org/x/tools
)

replace golang.org/x/net => example.com/fork/net v1.4.5

As a result, Go 1.18 comes with the useful Fuzzing and Workspaces features, along with the Generics feature that was discussed, talked and introduced more than a year. Apart from that, many improvements and fixes will also be in this version. You can follow the status of the version and what it contains via this link.