Forem: Lokesh Sanapalli

HTTP retry with exponential backoff in java spring boot

Lokesh Sanapalli — Sat, 25 Mar 2023 06:04:06 +0000

Implementing HTTP retry with exponential backoff in spring boot is not trivial. In this blog post, we will learn to implement the HTTP retry mechanism in java and spring boot.

Rest Template

Rest template is the popular HTTP client in spring boot. We will create a configuration and create a bean for it so that the spring boot’s container picks it up.

package com.abc.config;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.util.CollectionUtils;
import org.springframework.web.client.RestTemplate;

@Configuration
public class RestTemplateConfig {

    @Bean
    public RestTemplate restTemplate() {
        RestTemplate restTemplate =
                new RestTemplateBuilder().setConnectTimeout(Duration.ofSeconds(30))
                .setReadTimeout(Duration.ofSeconds(90)).build();
        List<ClientHttpRequestInterceptor> interceptors = restTemplate.getInterceptors();
        if (CollectionUtils.isEmpty(interceptors)) {
            interceptors = new ArrayList<>();
        }
        restTemplate.setInterceptors(interceptors);
        restTemplate.setRequestFactory(new HttpComponentsClientHttpRequestFactory());
        return restTemplate;
    }

}

Retry Config

We need to add spring-retry and spring-aspects to our dependencies.

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
    <version>1.2.5.RELEASE</version>
</dependency>

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-aspects</artifactId>
    <version>5.2.8.RELEASE</version>
</dependency>

package com.abc.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.backoff.BackOffPolicy;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.web.client.HttpServerErrorException;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RetryConfig {


    ExceptionClassifierBackoffPolicy exceptionClassifierBackoffPolicy;

    @Autowired
    public RetryConfig(ExceptionClassifierBackoffPolicy exceptionClassifierBackoffPolicy) {
        this.exceptionClassifierBackoffPolicy = exceptionClassifierBackoffPolicy;
    }

    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(5);
        ExponentialBackOffPolicy exponentialBackOffPolicy = new ExponentialBackOffPolicy();
        exponentialBackOffPolicy.setInitialInterval(5000L);
        exponentialBackOffPolicy.setMaxInterval(40000L);

        Map<Class<? extends Throwable>, BackOffPolicy> policyMap = new HashMap<>();
        policyMap.put(HttpServerErrorException.class, exponentialBackOffPolicy);
        exceptionClassifierBackoffPolicy.setPolicyMap(policyMap);
        retryTemplate.setRetryPolicy(simpleRetryPolicy);
        retryTemplate.setBackOffPolicy(exceptionClassifierBackoffPolicy);

        return retryTemplate;
    }
}

ExponentialBackoffPolicy takes the following parameters.

initialinterval - The initial number of seconds to wait.

maxInterval - The maximum number of seconds to wait.

multiplier - The factor to multiply the initial interval with.

We created a hashmap of the exception class and the policy to use and then we added it to our rest template. A point to note here is after setting the exponential backoff policy, the simple retry policy also needs to be set. If it is not set, by default only 3 retries will be made. So, in the above code, we set the max retires as 5 to the simple retry policy because, in the exponential back-off policy, we set 5 seconds as the initial delay. So, on the first retry, it waits for 5, then 10, 20, and 40. Totally 5 retries will be made. Make sure to adjust these numbers that works better for you.

ExceptionClassifierBackoffPolicy

There is a default ExceptionClassifierRetryPolicy using which we can configure policies for exception classes. But we cannot use ExponentialBackOffPolicy it. Because in spring retry, Retry and Backoff are two different policies implementing different interfaces. So, we need to create our own custom ExceptionClassifierBackoffPolicy class for it.

package com.abc.config;

import org.springframework.classify.Classifier;
import org.springframework.classify.ClassifierSupport;
import org.springframework.classify.SubclassClassifier;
import org.springframework.retry.RetryContext;
import org.springframework.retry.backoff.BackOffContext;
import org.springframework.retry.backoff.BackOffInterruptedException;
import org.springframework.retry.backoff.BackOffPolicy;
import org.springframework.retry.backoff.NoBackOffPolicy;

import java.util.HashMap;
import java.util.Map;

public class ExceptionClassifierBackoffPolicy implements BackOffPolicy {

    private static class ExceptionClassifierBackoffContext implements BackOffContext, BackOffPolicy {
        Classifier<Throwable, BackOffPolicy> exceptionClassifier;
        RetryContext retryContext;

        Map<BackOffPolicy, BackOffContext> contextMap = new HashMap<>();

        ExceptionClassifierBackoffContext(Classifier<Throwable, BackOffPolicy> exceptionClassifier, RetryContext retryContext) {
            this.exceptionClassifier = exceptionClassifier;
            this.retryContext = retryContext;
        }

        @Override
        public BackOffContext start(RetryContext context) {
            return (BackOffContext) context;
        }

        @Override
        public void backOff(BackOffContext backOffContext) throws BackOffInterruptedException {
            BackOffPolicy policy = exceptionClassifier.classify(retryContext.getLastThrowable());
            BackOffContext policyContext = contextMap.get(policy);
            if (policyContext == null) {
                policyContext = policy.start(retryContext);
                contextMap.put(policy, policyContext);
            }
            policy.backOff(policyContext);
        }
    }

    private Classifier<Throwable, BackOffPolicy> exceptionClassifier =
            new ClassifierSupport<>(new NoBackOffPolicy());

    public void setPolicyMap(Map<Class<? extends Throwable>, BackOffPolicy> policyMap) {
        this.exceptionClassifier = new SubclassClassifier<>(policyMap, new NoBackOffPolicy());
    }
    @Override
    public BackOffContext start(final RetryContext context) {
        return new ExceptionClassifierBackoffContext(exceptionClassifier, context);
    }

    @Override
    public void backOff(final BackOffContext backOffContext) throws BackOffInterruptedException {
        ExceptionClassifierBackoffContext classifierBackOffContext = (ExceptionClassifierBackoffContext) backOffContext;
        classifierBackOffContext.backOff(backOffContext);
    }
}

Main class

Now combining all of it, here’s the main class.

package com.abc;

import com.abc.config.ExceptionClassifierBackoffPolicy;
import com.abc.config.RestTemplateConfig;
import com.abc.config.RetryConfig;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;

public class main {
    public static void main(String[] args) {
        RetryConfig retryConfig = new RetryConfig(new ExceptionClassifierBackoffPolicy());
        RestTemplateConfig restTemplate = new RestTemplateConfig();
        HttpEntity<String> resp = new HttpEntity<>(null);
        ResponseEntity<String> newResp =
                retryConfig.retryTemplate().execute(context -> restTemplate.restTemplate().exchange(
                        "https://abc-test" +
                                ".free" +
                                ".beeceptor" +
                                ".com/test/abc",
                        HttpMethod.GET,
                        resp,
                        String.class));
        System.out.println(newResp.getBody());
    }
}

How I built domain hunter using OpenAI in a weekend

Lokesh Sanapalli — Sat, 25 Mar 2023 05:55:10 +0000

Introduction

When ChatGPT was released, everyone was in awe, and a few people built production on top of it. I was also in the same hype and created a product. It is called domain hunter — A tool that generates domains that are available for registration using OpenAI. You give terms in the search bar and it generates the domains.

System Design

Tech Stack

Python
Django
Celery
Redis
PostgreSQL
TailwindCSS
jQuery

As explained in the above diagram, the user enters the search terms, they will be put in a rabbitmq queue. They will be picked up by celery workers and call OpenAI for domain suggestions. At this point, we don’t know whether all the domains suggested by OpenAI are available or not. So, we need to call domain APIs to get the availability and price.

There are other sets of workers which consume messages from rabbitmq and call domain APIs. Then they put that data in redis and the database. So, the REST API controller will be polling redis for new data. Whenever it is found, it will return to the client.

Database Design

core_searchterm is the table that stores the terms and max_tokens to be used by OpenAI.
core_apiprovider contains the configuration of the API provider. It contains the base URL, token, etc…
core_apicallaudit contains other requests and responses of every API call. It has a foreign key to the API provider and search term.
core_domainsuggestion contains a foreign key to the search term and its available domain details.
core_visitor_search_terms and users_user_search_terms maintains the mapping of user/visitor to search terms.

End Notes

Currently, it has a free and paid plan. I released it 3 weeks ago. So far I did not see a single user sign up. People are visiting the website, searching, and disappearing. It seems I did not find the market fit yet. I am still trying, let’s see.

How to deploy django application to production

Lokesh Sanapalli — Sat, 25 Mar 2023 05:46:22 +0000

Django is a popular web framework with all batteries included. In this blog post, we will learn how to deploy it to production

Introduction

If you are looking for a guide to deploying a Django application to production, then you are at the right place. This guide assumes that you have a working Django application. You can check if your application is healthy by running this command python manage.py check .

Components

Gunicorn

python manage.py runserver is not ideal for production. We need to run it with gunicorn and wsgi. We can also use gevent workers to take advantage of multiple processors. Remember python’s GIL that makes it hard to achieve concurrency?

#!/bin/bash

NAME="ProjectName"
DJANGODIR=/root/project_name
USER=root
GROUP=root
NUM_WORKERS=5
DJANGO_SETTINGS_MODULE=config.settings.production
DJANGO_WSGI_MODULE=config.wsgi
echo "Starting $NAME as `whoami`"

cd $DJANGODIR
exec venv/bin/python venv/bin/gunicorn ${DJANGO_WSGI_MODULE}:application --name ${NAME} --workers ${NUM_WORKERS} --worker-class gevent --bind unix:///tmp/gunicorn.sock --log-level info --access-logfile ${DJANGODIR}/logs/django_server_stdout.log --error-logfile ${DJANGODIR}/logs/django_server_stderr.log --timeout 300

Supervisor

Supervisor is a utility to manage multiple processes. It is like pm2 in node.js. We need to add gunicorn and celery worker if you have any.

[program:gunicorn]
directory = /root/project_name
command = /bin/bash -c "/bin/bash gunicorn_start"
user = root
stdout_logfile = /root/project_name/logs/gunicorn_stdout.log
stderr_logfile = /root/project_name/logs/gunicorn_stderr.log
redirect_stderr = false
environment = LANG=en_US.UTF-8,LC_ALL=en_US.UTF-8

[program:celery_worker]
command = /bin/bash -c "/root/project_name/venv/bin/celery -A config.celery worker -c 10 -P gevent -Q openai -n openai_w1@%%h"
directory = /root/project_name
user = root
stdout_logfile = /root/project_name/logs/openai_worker_stdout.log
stderr_logfile = /root/project_name/logs/openai_worker_stderr.log
redirect_stderr = false
autostart = False
environment = LANG=en_US.UTF-8,LC_ALL=en_US.UTF-8

Nginx

We need a reverse proxy to route all the requests to the web server.

server {
    server_name project_name.com;

    location = /favicon.ico { access_log off; log_not_found off; }
    client_max_body_size 4G;
    access_log /root/project_name/logs/nginx_access.log;
    error_log /root/project_name/logs/nginx_error.log;
    location /static/ {
        autoindex on;
        alias /root/project_name/staticfles/;
    }

    location /media/ {
        autoindex on;
        alias /root/project_name/media/;
    }

    location / {
        proxy_redirect off;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_pass http://unix:/tmp/gunicorn.sock;
    }

    listen 80;
}

server {

    if ($host = www.project_name.com) {
        return 301 http://$host$request_uri;
    }

    server_name www.domain.com;
    listen 80;
    return 404;
}

Glue them All

# OS dependencies
sudo apt-get install software-properties-common
sudo apt-add-repository universe
sudo apt-get update
sudo apt-get upgrade
sudo apt-get install -y gcc libpq-dev make build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev wget curl llvm libncurses5-dev libncursesw5-dev xz-utils tk-dev libffi-dev liblzma-dev python-openssl git python3 python-dev python3-dev libxml2-dev libxslt1-dev zlib1g-dev python-pip libmysqlclient-dev gcc libpq-dev python-dev python-pip python-wheel python3-dev python3-pip python3-venv python3-wheel
# install nginx and supervisor
sudo apt-get install nginx supervisor

By default, nginx runs with the user www-data and other processes run on root or a different user. So, all the static files give 403 forbidden because of it. To fix this, we need to create a new group, add www-data and root to it and add it to the permissions of static files.

sudo groupadd nginx
sudo usermod -a -G nginx root
sudo usermod -a -G nginx www-data
sudo chown root:nginx /root
# assuming that your project is in root
sudo chown root:nginx /root/project_name
sudo chown -R root:nginx /root/project_name/projectfiles
sudo chmod 755 /root
sudo chmod 770 /root/project_name
sudo chmod 770 -R /root/project_name/staticfiles

Copy the above supervisor config to /etc/supervisor/conf.d/django.conf and nginx config to /etc/nginx/sites-available/project_name.conf using the below command.

cp django.conf /etc/supervisor/conf.d/django.conf
cp project_name.conf /etc/nginx/sites-available/project_name.conf
ln -s /etc/nginx/sites-available/project_name.conf /etc/nginx/sites-enabled/project_name.conf

Start gunicorn and nginx

supervisorctl start gunicorn celery_worker
sudo service nginx restart

If you are using Ubuntu, check if the firewall is enabled using ufw status command. Allow ports 80 and 443 in the firewall using ufw allow 80 and ufw allow 443 . Many times I used to wonder why the website is inaccessible but accessible inside the machine. It is because those ports are blocked by the firewall.

Redeployment Script

To redeploy on every change, it becomes cumbersome to execute every command. Here is the shell script to execute.

source venv/bin/activate
source .envrc
git pull origin main
python manage.py check
python manage.py migrate
npm run build
python manage.py collectstatic --no-input
supervisorctl -c /etc/supervisor/supervisord.conf restart gunicorn celery_worker
sudo service restart nginx

HTTP retry and logging using python requests

Lokesh Sanapalli — Wed, 01 Mar 2023 09:21:01 +0000

HTTP retry and logging using python requests

Introduction

Python’s requests library is a popular library for making HTTP requests. It is a wrapper on python’s built-in urllib module. It gives a clean API with many features. In this post, we will see how to implement retry and logging when using the requests library.

Requests Design

requests is designed modularly using different object oriented patterns. All the methods we use such as requests.get , requests.post etc.. are present api.py which simply creates a session object and sends the request. The Session class present in session.py is the core and heart. It composes HTTPAdapter from urllib3 which is the low-level module in python for sending HTTP requests. It also composes PreparedRequest in models.py which does the preparation of request such as sanitizing input, request body, params etc…

Logging

Why logging is important? it’s a no-brainer! it will be very helpful when troubleshooting issues in a production environment where you can’t debug the program. When building distributed systems, if your application communicates with multiple systems then adding logging to HTTP requests is a must. While logging the request, you can send a unique request id in the headers which helps in tracing the whole lifecycle of the request throughout the system — This is called distributed tracing — I will write a separate blog post on this. Let’s see how can we add logs to python’s requests.

Change the log level of the underlying urllib3 module.

urllib3 provides a handy function called add_stderr_logger which adds a stream handler and sets the level DEBUG by default. We can see the code here for the same.

import requests
import urllib3

urllib3.add_stderr_logger()

resp = requests.get(
    "https://jsonplaceholder.typicode.com/posts",
    headers={"X-Request-ID": "abcd123"},
    params={"foo": "bar"},
    timeout=(10, 30),
)

The above code prints the following

2023-02-26 23:35:32,110 DEBUG Added a stderr logging handler to logger: urllib3
2023-02-26 23:35:32,166 DEBUG Starting new HTTPS connection (1): jsonplaceholder.typicode.com:443
2023-02-26 23:35:32,352 DEBUG https://jsonplaceholder.typicode.com:443 "GET /posts?foo=bar HTTP/1.1" 200 None

This does not log the request headers and body which we may require. So, the only alternative way is to manually log the request and response as we desire. Let’s write a decorator for it.

inherit requests.Session

Just like other HTTP libraries (Axios in javascript), requests do provide hooks or interceptors. A hook is a function that will be called when an event has happened. Unfortunately, requests do not have hooks for the request . It only has for response.

The only way to add logging is to extend requests.Session class and add logging.

class RequestSession(requests.Session):

    def request(
        self,
        method,
        url,
        params=None,
        data=None,
        headers=None,
        cookies=None,
        files=None,
        auth=None,
        timeout=None,
        allow_redirects=True,
        proxies=None,
        hooks=None,
        stream=None,
        verify=None,
        cert=None,
        json=None,
    ):
        logger.debug(f"Before making HTTP request. url={url} method={method}"
                     f"params={params} data={data}"
                     f"headers={headers} json={json} timeout={timeout}")
        start_time = time.time()
        response = super().request(
            method,
            url,
            params=params,
            data=data,
            headers=headers,
            cookies=cookies,
            files=files,
            auth=auth,
            timeout=timeout,
            allow_redirects=allow_redirects,
            proxies=proxies,
            hooks=hooks,
            stream=stream,
            verify=verify,
            cert=cert,
            json=json,
        )
        logger.debug(f"After making HTTP request with params url={url} method={method}"
                     f"params={params} data={data} headers={headers}"
                     f"json={json} timeout={timeout} response={response.text}"
                     f"response_status_code={response.status_code}"
                     f"response_headers={response.headers} total_time_taken={time.time() - start_time} seconds")
        try:
            response.raise_for_status()
        except requests.HTTPError:
            logger.critical(f"HTTP request failed. url={url} method={method}"
                            f"params={params} data={data} headers={headers}"
                            f"json={json} response={response.text}"
                            f"response_status_code={response.status_code}")
        return response

If required, we can always add more details to the logs. Also, make sure to configure the logging level of the module. The below code sets the log levels.

import logging
import requests
import sys
import time


logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

if __name__ == "__main__":
    with RequestSession() as session:
        resp = session.get("https://jsonplaceholder.typicode.com/posts", headers={"X-Request-ID": "abcd123"},
                           params={"foo": "bar"},
                           data={"a": 1, "b": 2}, timeout=(10, 30)
                           )

Retry

Having retries with exponential backoff is a must in a distributed system to make systems more resilient and fault-tolerant. Let’s see how we can implement retry mechanism with python requests.

The retry mechanism is provided by urllib3 package which is the underlying module of requests.

import requests
from requests.adapters import Retry, HTTPAdapter

retry = Retry(total=10, connect=5, read=5, allowed_methods=['GET', 'POST', 'PUT'], backoff_factor=10, status_forcelist=[500, 502, 503, 504])
session = requests.Session()
session.mount("http://", HTTPAdapter(max_retries=retry))
resp = session.get("https://jsonplaceholder.typicode.com/posts")
print(resp.json())

total - the total number of requests to make in total. This argument takes higher precedence which means if you set the total to 10 and other counts higher, the total will be taken.

connect - is the number of retries to make if there is an exception in connecting to the target host.

read - is the number of retries to make if after connecting to the host, if there is any exception in reading data.

allowed_method - Be careful with this. By default, urllib3 includes only idempotent verbs. According to HTTP RFC, GET, PUT, DELETE are the idempotent verbs. When using this, make sure all your verbs are idempotent.

backoff_factor - Number of seconds to wait between two retries. {backoff factor} (2 * ({number of total retries} - 1)) - is the formula to compute backoff time. If backoff_factor is 10, then 5s, 10s, 20s, 40s etc…

status_forcelist - It is the list of status codes on which retries are to be made.

How Retry is Implemented in urllib3

Read this only if you are interested to learn how retry in urllib3 is implemented. In urllib3, connectionpool.py is the core and low-level module for sending HTTP requests. In the function urlopen, if there is any exception then retries.increment is called followed by retries.sleep

retries = retries.increment(method, url, error=new_e, _pool=self, _stacktrace=sys.exc_info()[2])
retries.sleep()

The function increment in Retry increments the retry count, checks if the retry limit is exhausted. If yes, then it raises an exception. The code goes on like below.

history = self.history + (
    RequestHistory(method, url, error, status, redirect_location),
)

new_retry = self.new(
    total=total,
    connect=connect,
    read=read,
    redirect=redirect,
    status=status_count,
    other=other,
    history=history,
)

if new_retry.is_exhausted():
    reason = error or ResponseError(cause)
    raise MaxRetryError(_pool, url, reason) from reason

Now let’s see what rety.sleep function do. It checks if the response contain the header Retry-After and respects that. If it’s not present, then it gets the backoff time and sleeps.

def sleep_for_retry(self, response=None):
    retry_after = self.get_retry_after(response)
    if retry_after:
        time.sleep(retry_after)
        return True
    return False

def _sleep_backoff(self):
    backoff = self.get_backoff_time()
    if backoff <= 0:
        return
    time.sleep(backoff)

def sleep(self, response=None):
    if self.respect_retry_after_header and response:
        slept = self.sleep_for_retry(response)
        if slept:
            return
    self._sleep_backoff()

For the whole code snippet, please find here.

Caching patterns

Lokesh Sanapalli — Mon, 13 Feb 2023 03:18:40 +0000

Introduction

How many heard the phrase "Cache invalidation is the toughest problem in computer science?" I was first introduced to the term "cache" in my computer architecture class in engineering. When I first learned, I did not understand the concept. It took me some years to wholly understood the concept. In this blog post, we will look at the basics of caching and different strategies.

Basics

What is caching? caching is a mechanism of storing data in fast and secondary storage temporarily. The main goal of caching is to decrease the response time and increase speed. The fundamental concept is simple, primary storage gives persistence and durability that stores the data permanently. But, they are slow. We need secondary storage that is fast but temporary. That's what the RAMs are designed for. They are fast but the data stored in them is temporary. We also call RAM in-memory. Let's try to understand the mechanism with the diagram below.

The above diagram is a no-brainer. We first check in the cache because it's fast, if not found fetch it from primary storage and write it back to the cache.

Caching is implemented in different use cases. A few of them are

Disk page caching: Hard disks are cylindrical structures divided into sectors, tracks, and pages. A page is a continuous block of memory. Accessing the hard disk every time requires an I/O overhead and CPU context switching. To optimize this, the UNIX operating system caches the pages in the main memory.
Web server caching: Web servers like nginx caches the frequently accessed pages to increase speed and responsiveness. They simply store the entire HTML content in the main memory and serve from there.
Application level caching: Applications like web browsers, code editors and your backend web APIs cache the result in in-memory.

Cache Eviction Policies

The capacity of our RAM is of some fixed size. So, when it reaches its size, it needs to omit some data. This process is called eviction. Below are different eviction techniques used.

Least Recently Used

As the name says, remove the least recently used entry. A doubly linked list data structure is used commonly to implement this mechanism. The data is stored in the sorted order of access where the most recently used at the beginning and the least recently used at the end. Every time a cache key is accessed, it will be put at the front.

Least Frequently Used

A frequency count of each cache key will be maintained. Every time a key is accessed, the frequency count will be updated. Whenever the cache reaches its size, the less frequent item will be removed.

Caching Patterns

There are different patterns of caching. We will look at each of them.

Cache Aside

This is the most common pattern. We already discussed this in the above flow diagram. We will first check in the cache, and if not found, read from the database and write to the cache again.

Read Through

In cache through pattern, the application server asks the cache for the data, in case of a cache miss, the cache itself fetches from the database and saves in the cache. In order to implement this pattern, the caching software should have libraries to integrate with the database technology. Very few caching software implement this pattern. An example is AWS DAX which integrates with DynamoDB.

Write Behind

In this pattern, the writes to the cache happen asynchronously through the message queue. The advantage of this pattern is latency is reduced as the writes to the cache are not happening synchronously. But, the downside of this pattern is cache misses increase if there is a lag in writing data to the cache.

Database Migrations in Django

Lokesh Sanapalli — Wed, 08 Feb 2023 18:03:34 +0000

Introduction

Django gives a powerful framework to manage database migrations. Without this, managing schema and data changes would be hectic work. Every schema change to the database can be managed through Django. You make a change to your Django model, create a migration and apply that migration to the database. If you want to reverse it, you can do that too. All this is irrespective of the underlying database technology. How simple, beautiful, and declarative isn't it? do you feel it? I can feel it. In this blog post, we will learn how Django migrations work under the hood, and Django's data migrations

Why Migrations Are Required?

So, why are migrations required? what will happen without it? Let's say you are working on a Django application, and you make changes to your model but they are not synced to the database.

Track SQL queries manually: You need to execute them manually. But, what if you missed executing a query in between and caused a whole chaos? So, you need to track them in some place in a sequential manner. Where do you store it? maybe create a folder and add all .sql files there.
Deploy in multiple environments: Assume you want to deploy your app on multiple environments such as dev, QA, staging and production, etc... So, you need to execute all these queries manually on each machine in sequential order.

Django's migrations framework solved all the above problems and made it so convenient for users to manage them.

Basics

In the below code, we have added a new field called about which is a type of TextField . So, After making this change in our models, we need to propagate this change to the database.

We need to create a migration and apply it. The commands are

python manage.py makemigrations # creates a migration file in `migrations` folder
python manage.py migrate # applies the migration to the database by exeucting SQL query

makemigrations command creates a migration file in migrations the folder. migrate the command applies that migration to the database through SQL. If we take a look at the migration file, it looks like this

from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('job', '0002_jobapplication'),
    ]

    operations = [
        migrations.AddField(
            model_name='company',
            name='about',
            field=models.TextField(
                    _("About Company"),
                    help_text=_(
                        "Write about the company. It will be displayed on the web page to the users."
                    ),
                  ),
    ]

Let's dissect the above code.

Migration Class: Migration is a class inherited migrations.Migration base class which has the mechanism to apply migrations.

Dependencies: It has an array dependencies where each item is a tuple of the app name and migration file name. It means that in order for this migration to be executed, the dependencies should be executed first.

Why do dependencies are required? Imagine there are no dependencies, say you have a migration to delete or alter a field. Django connects to the database and tries to execute the SQL query but it will fail with field `xyz` does not exist error. Because you may have not run the query to add that column. So, dependencies are important to maintain the consistency of the database.

Operations: is an array of operations. Adding a column, modifying a column and deleting a column, Adding a new table are common operations. Django's migration framework has these methods for each operation. Under the hood, those methods have the logic to generate SQL queries.

Django Migrations 🤝 Graph Data Structure

In order for Django to apply migrations, it needs to know already applied migrations. So, it needs to store the state somewhere. Django stores the data in a table called django_migrations table with app , name and applied timestamp. Every time a migration is applied, it adds a new entry to the table in order. Also, Django uses the graph data structure to check the consistency of the migrations. It builds the graph of applied migrations and their dependencies (remember? we give dependencies for each migration) where each node is the migration name and the directed edge is the dependency. It checks for any conflicts and raises an error if any.

Conflicting migrations detected; multiple leaf nodes in the migration graph

The above error is a common error while applying migrations. It occurs when an app has multiple leaf nodes. This commonly happens when you are working in a team and multiple people create the migration simultaneously. Refer to the below diagram.

Assume that migration "22" is already applied, Person A had added a migration that adds location to the user model and Person B adds an avatar. Both of the migrations have "22" as the dependency. When Django tries to apply a migration, it doesn't know which one to apply first. What if one migration deletes a field and another migration alters the same field? So, it is impossible to choose a migration. So, that's how the conflict occurs for Django. How does this conflict get resolved? Simple, we need to merge them using the below command and create a single migration.

python manage.py makemigrations --merge

Data Migrations

Along with database schema changes, we can create migrations for data too. First, we need to create a migration file with the below command.

python manage.py makemigrations job --empty --name load_job_data

We are creating an empty data migration which looks like below

# Generated by Django 3.2.10 on 2023-02-08 17:39

from django.db import migrations


class Migration(migrations.Migration):

    dependencies = [
        ('job', '0004_alter_job_location'),
    ]

    operations = [
    ]

We need to add our custom function to seed data in the operations list. The function's name can be anything but it should have the following structure. We need to use Django's app registry API to access the models and operate on them as we see below.

def load_job_data(apps, schema_editor):
    Job = apps.get_model("job", "Job")
    resp = requests.get("https://jobs.abc.com/api/v1/jobs")
    for datum in resp.data:
        Job.objects.create(url=datum.url, location=datum.location)

Decorators in python (part-2)

Lokesh Sanapalli — Sun, 05 Feb 2023 06:11:53 +0000

Introduction

If you have not read the previous post, please go through it because it is a prerequisite for this. In this post, we will go into some more depth and learn about decorators with arguments.

functools.wraps decorator

We learned from the previous blog post that when a function is decorated with a decorator, the representation of it will be changed to decorator.<locals>.wrapper at 0x109e233a0> which looks ugly. How can we keep the original string representation of the function? Here comes functools.wraps the decorator. It is a decorator — again, so many decorators huh? — which restores the original function name. We will use it like the one below.

import functools

def decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        pass
    return wrapper

In a nutshell, what it does is simply sets __name__ , __doc__ and __module__ properties of the wrapper function to the original one.

wrapper.__name__ = func.__name__
wrapper.__doc__ = func.__doc__
wrapper.__module__ = func.__module__

Decorators with arguments

First, let's write a retry decorator which makes an API call and retries on failure.

import requests
import time
import logging


logger = logging.getLogger(__name__)

def retry(make_http_call):
    def wrapper(*args, **kwargs):
        curr_retry_count = 0
        max_retries = 5
        retry_interval = 10
        while init_retry_count <= max_retries:
            try:
                response = make_http_call(*args, **kwargs)
                if response.status_code in [200, 201]:
                    return response
            except requests.RequestException:
                logger.error("error in making call")
            time.sleep(retry_interval * (math.pow(2, curr_retry_count - 1))) # retry_interval * 2^(n-1) -> exponential backoff
            curr_retry_count += 1
        raise requests.RequestException("problem in making HTTP call please check logs")
    return wrapper

In the above code snippet, retry is a function accepting make_http_call as a parameter. Assume that make_http_call triggers an HTTP request. The code is a no-brainer, it makes an HTTP request in a loop until the request is a success or the maximum retries are reached. It waits for some time until the next retry i.e. exponential backoff.

We want to make retry customizable by accepting max_retry_count and retry_interval because they may vary depending upon the usecase. How do we do that? simple, we will create another function which takes two parameters and simply wrap the above function inside it.

import requests
import time
import logging


logger = logging.getLogger(__name__)


def retry(max_retries, retry_interval):
    def inner(make_http_call):
        def wrapper(*args, **kwargs):
            curr_retry_count = 0
            while init_retry_count <= max_retries:
                try:
                    response = make_http_call(*args, **kwargs)
                    if response.status_code in [200, 201]:
                        return response
                except requests.RequestException:
                    logger.error("error in making call")
                time.sleep(retry_interval * (math.pow(2, curr_retry_count - 1))) # retry_interval * 2^(n-1) -> exponential backoff
                curr_retry_count += 1
            raise requests.RequestException("problem in making HTTP call please check logs")
        return wrapper
    return inner


@retry(10, 60)
def make_http_request(url, params=None, data=None):
    pass

Mathematically, the above function make_http_request becomes make_http_request = retry(10, 60)(make_http_request) and calling that function would be like

make_http_request(
    "https://api.twitter.com/v2/users",
    params={"username": "lsanapalli"}
) = retry(10, 60)(make_http_request)(
    "https://api.twitter.com/v2/users",
    params={"username": "lsanapalli"}
)

Using multiple decorators

Can we decorate a function with multiple decorators? yes. What will be the order of execution? simple, let's consider an example and stick to our basics.

@decorator3
@decorator2
@decorator1
def myfunc(a, b):
    pass

Consider we have a function myfunc decorated with 3 decorators as above. Firstly, myfunc will be passed as an argument to decorator1 and the result of that will be passed to decorator2 and so on...

myfunc = decorator3(decorator2(decorator1(myfunc)))

If you prefer a video version, I made a video on the same topic.

Decorators in python (part-1)

Lokesh Sanapalli — Sun, 05 Feb 2023 06:11:37 +0000

Introduction

As usual, let's stick to the first principles. Let's start with what was taught in our schools. What is a function in mathematics? f(x) -> y spelled as f of x is a function that takes x as input and returns y as output. Right, It's the same thing in programming. A function takes some parameters as input and returns a value or object as output.

Let's understand the difference between the definition and the calling of a function.

 def make_http_request(url: str) -> http.Response:
    resp = http.request(url)
    if resp.status_code == 200:
        return resp
    return None

In the above code snippet, we defined a function make_http_request that takes url as the "parameter" and returns http.Response the object as output. As the definition we mean, the function will be registered with in-memory (RAM) when the program run. The calling of a function happens when we supply the "arguments" to the function as shown in the below code.

resp = make_http_request("https://twitter.com/api/users")
print(resp)

notice the difference between parameter and argument. "parameters" are the variables passed to the function, they are not the real values. "arguments" are the real values passed to the function. It's not much important to differentiate them as some people use both words interchangeably.

Function Basics

Functions are first-class citizens in python. It means that they can be assigned to a variable, passed as an argument to a function, and returned from another function.

Assign function to a variable

def fetch_movies_by_actor(actor_name, movie_db):
    if actor_name == '' or actor_name is None:
        raise ValueError("actor name shouldn't be empty'")
    return filter(lambda movie: actor_name in movie["actors"], movie_db)

if __name__ == "__main__":
    obj = fetch_movies_by_actor
'''
>>> print(obj)
<function fetch_movies_by_actor at 0x109f01280>
>>> print(fetch_movies_by_actor)
<function fetch_movies_by_actor at 0x109f01280>
'''

As seen from the above snippet, we defined a function named fetch_movies_by_actor and assigned it to a variable called obj. When I print both, they both printed the same.

Pass as an argument

def fetch_movies_by_actor(actor_name, movie_db):
    if actor_name == '' or actor_name is None:
        raise ValueError("actor name shouldn't be empty'")
    return filter(lambda movie: actor_name in movie["actors"], movie_db)

def fetch_movie_details(actor_name, func):
    if func is not None:
        print(func)
        movies = func(actor_name)
        return movies
    return []

if __name__ == "__main__":
    result = fetch_movie_details("Tom Cruise", fetch_movies_by_actor)

As seen in the above example, fetch_movies_by_actor the function is passed as an argument to fetch_movie_details and it is called there.

Returning from another function

def fetch_movie_genres_by_year(year):
    filtered_movies = list(filter(lambda movie: movie["year"] == year, movie_db))
    def get_genres_by_movie_name(movie_name):
        movie = list(filter(lambda movie: movie["name"] == movie_name ,filtered_movies))[0]
        return movie["genre"]
    return get_genres_by_movie_name

As seen in the above code snippet, the function fetch_movie_genres_by_year returned another function get_genres_by_movie_name . To obtain the final result, we need to do this 👇.

movie_genres = fetch_movie_genres_by_year(2009)("XYZ")

Let's try to understand what's happening there. In mathematical terms, f(x) -> g(x) -> result the function f(x) returned g(x) which returned the result. So, to get the final result, we need to call f(x) and then g(x) that's what we did above.

Decorators

What is a decorator? A decorator is a function, which takes a function as a parameter and returns another function. It may be slightly confusing at this point, but everything will be clear at the end. Why do decorators exist? What are its uses? Decorators extend the functionality of a function which reduces the lines of code to write. When to use decorators? Simple, when you see a lot of repetitive code.

Let's say you have hundreds of functions defined and you want to log the parameters of the function and the result of it for troubleshooting purposes. Using decorators, you can add them in a single place and apply them to all the functions.

def decorator(func): # decorator is a function, taking another function "func" as a parameter
    def wrapper(*args, **kwargs):
        # do something before evaluating function
        result = func(*args, **kwargs)
        # do something with result or some other stuff
        return result
    return wrapper # decorator is returning another function

@decorator
def myfunc(a, b):
    # function body
    pass
'''
myfunc's identity will be changed as myfunc = decorator(myfunc)
'''

In the above code snippet, myfunc is decorated with the decorator named decorator using @ symbol which is syntactic sugar to use decorators in python. So, What happens when you decorate a function? Let's understand it.

@decorator
def myfunc(a, b):
    # function body
    pass

def myfunc2(a, b):
    # function body
    pass

if __name__ == "__main__":
    print(myfunc)
    print(myfunc2)

'''
>>> print(myfunc)
<function decorator.<locals>.wrapper at 0x109e233a0>
>>> print(myfunc2)
<function myfunc2 at 0x109e23430>
'''

As seen in the above code snippet, myfunc is printed with the name of the decorator and the function returned by the decorator i.e. wrapper. Whereas myfunc2 is printed as usual.

It is clear that when a function is decorated with a decorator, the original function's definition will change to myfunc = decorator(myfunc) i.e. myfunc = decorator(myfunc) -> wrapper . If we look at the decorator's definition, the func parameter is nothing but the original function which is decorated.

Calling the original function will be equal to myfunc(*args, **kwargs) = decorator(myfunc)(*args, **kwargs) i.e. myfunc(*args, **kwargs) = decorator(myfunc) -> wrapper(*args, **kwargs). If we look at the decorator definition again, the arguments passed to the wrapper function inside the decorator are nothing but the arguments passed to the original function. Also, be sure to call the original function in the inner function of the decorator else the whole purpose of the decorator will be lost.

Logger Example

import logging


logger = logging.getLogger(__name__)

class CustomException(Exception):
    pass

def exception_logger(func):
    """decorator to log the exceptions"""

    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as exc:
            logger.exception(
                "Unhandled exception in function %s , exception is %s", func.__name__, exc
            )
            raise CustomException("Unhandled exception in function %s" % func.__name__) from exc

    return wrapper

In the above code snippet, the original function is called in between try and except block where we are catching and logging the exceptions.

Let's use the decorator and see what happens.

@exception_logger
def func(a, b): # func = exception_logger(func)
    return a / b

print(func(1, 0)) # func(1, 0) = exception_logger(func)(1,0)

'''
Unhandled exception in function func , exception is division by zero
Traceback (most recent call last):
  File "<ipython-input-1-5525c8c2a27e>", line 14, in inner
    return func(*args, **kwargs)
  File "<ipython-input-3-e825ef26bd9c>", line 3, in func
    return a / b
ZeroDivisionError: division by zero
'''

As we see from the above snippet, the exception was logged by the decorator. When the program runs, firstly the func will be changed to func = exception_logger(func) . When we call the original function, we are indirectly calling the wrapper function of the decorator.

Timing Example

Let's write a decorator to measure the time taken by a function. Below is the code snippet for it. It is self-explanatory.

import functools
import time


def timeit(func):
    def wrapper(*args, **kwargs):
        before_calling = time.time()
        result = func(*args, **kwargs)
        after_calling = time.time()
        print("function %s has took %s seconds" % (func.__name__, (after_calling - before_calling)))
        return result
    return wrapper

If you prefer a video version, check out 👇

Change Data Capture Using Debezium

Lokesh Sanapalli — Sun, 05 Feb 2023 06:00:07 +0000

Introduction

What is change data capture? A change that occurred in one data source will be captured and replicated in another data source. Let's understand it with a picture.

A: Databases maintain transactional log where it writes all DDL & DML queries. MySQL maintains binlog and PostgreSQL maintains WAL (write-ahead logs).

B & C: The source connector connects to the database and reads the transaction logs. It then prepares an in-memory structure of the change and writes to the message broker. Generally, CDC systems maintain a single schema for the change so that different database connectors transform the data into it.

D & E: The sink connector connects to the message broker, reads the data, and writes the change to the target database.

Use-cases of CDC

Let's think about different use cases of CDC in software engineering.

OLAP (online analytical processing) systems use CDC to migrate data from transactional databases to analytical databases.
OLTP (online transactional processing) systems can also use CDC as an event bus to replicate data in a different data store. For example, from MySQL to Elasticsearch.

One of the popular and widely-used systems is debezium. In this blog post, we will discuss how to set up debezium and how it works internally.

Debezium

Debezium works on top of kafka connect. Kafka Connect is a framework for streaming data between multiple systems. When deployed, it provides REST APIs to manage connectors. There are two connectors, the source, and the sink. As we saw from the above diagram, the source connector reads the source database and writes the data to the kafka topics. The sink connectors read from those kafka topics and write to the target database. Debezium is built using the kafka connect framework. It comes with connectors for different databases.

Debezium can only be installed through docker. The installation is very easy. Just pull the image from the registry and run it. More instructions are given here. Once deployed, the debezium server will be running on port 8083 by default. We can hit its REST API to create connectors.

# view connectors
curl -i -X GET localhost:8083/connectors
# create a connector
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d @mysql-connector.json
# delete a connector
curl -i -X DELETE localhost:8083/connectors/<connector name>

Debezium and MySQL

MySQL maintains a binary log (binlog) that writes all the schema changes and data changes in the same order as they happened. The debezium MySQL connector reads the binlog and produces change events for every schema and row-level change.

Create a MySQL User

In order for debezium to read binlog, it should have a user with specific permissions. Before we deploy our MySQL connector, let's create a user with the following commands.

CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';
GRANT SELECT, RELOAD, SHOW DATABASES, LOCK TABLES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';
FLUSH PRIVILEGES;

Enable binlogs

The procedure for enabling binlogs in a self-hosted environment is easy. The instructions are given here. If you are using managed services like AWS, you can enable it from the console. Also, AWS RDS gives a few stored procedures to set this.

# show rds config
call mysql.rds_show_configuration;
# set binlog retention period
call mysql.rds_set_configuration('binlog retention hours', 96);

Deploy MySQL Connector

Before deploying a MySQL connector, we need to prepare the JSON.

{
  "name": "debezium-connector1",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "abc123",
    "database.server.id": "130",
    "topic.prefix": "prod-debezium",
    "database.include.list": "mydatabase",
    "table.include.list": "mydatabase.users,mydatabase.orders",
    "snapshot.mode": "initial",
    "schema.history.internal.kafka.bootstrap.servers": "localhost:9092,locahost:9093,localhost:9094",
    "schema.history.internal.kafka.topic": "dbhistory.debezium",
    "provide.transaction.metadata": "true",
    "topic.creation.default.replication.factor": 2,
    "topic.creation.default.partitions": 10,
    "database.history.skip.unparseable.ddl": "true",
    "event.processing.failure.handling.mode": "warn",
    "snapshot.lock.timeout.ms": 180000,
  }
}

Let's try to understand the meaning of each configuration. I skip self-explanatory configs and mention only the important ones.

connector.class - This is the path to the connector in the debezium code. When this connector is deployed, that particular class will run and process the data. So, it is important to give this config correctly.

database.server.id - The unique identifier of the connector. This differentiates the other MySQL connector.

topic.prefix - In older versions, it used to be database.server.name . This provides a namespace for the current instance of the connector. Debezium creates kafka topics by appending this prefix with the table name i.e. <topic.prefix>.<table_name>. This should be unique for a connector else debezium assumes it is an existing connector and tries to resume from where it stopped.

Database History Topic

A client connects to the database and reads its current schema. But the schema can change anytime. So, the debezium connector cannot just read the current schema because it may be processing the older change events from the database. It needs to correlate the current change with the exact schema. So, It needs to store the schema changes somewhere.

We know that MySQL emits the schema changes along with the row level changes to the binlog. Debezium reads the DDL statements, prepares an in-memory representation of the table, and also stores them in a separate topic along with the binlog position. This topic is called database history topic. It is configured with schema.history.internal.kafka.topic .

When the connector restarts, it reads the schema history topic and starts rebuilding the in-memory representation of each table, and resumes reading the binlog where it left off. The debezium connector emits the schema along with the row-level change so that the consumer systems can rebuild the whole table.

A very important note about the history topic is it is only for debezium internal use and it should not be partitioned. Meaning, it should have a partition of only one. Why? because in kafka the ordering of messages is guaranteed only at the partition level, not at the topic level. So, if it is partitioned, then the order of schema changes will be mixed which leads to chaos.

This is how the data from the schema history topic looks

{
  "source" : {
    "server" : "prod-debezium"
  },
  "position" : {
    "transaction_id" : null,
    "ts_sec" : 1674648853,
    "file" : "mysql-bin-changelog.106869",
    "pos" : 69735280,
    "server_id" : 1229362390
  },
  "databaseName" : "licious",
  "ddl" : "CREATE TABLE `user_groups` (\n\t`id` bigint(20) NOT NULL AUTO_INCREMENT PRIMARY KEY,\n\t`user_id` bigint(20),\n\t`customer_key` varchar(25) NOT NULL,\n\tCONSTRAINT FOREIGN KEY `fk_user_id` (user_id) REFERENCES `users`(id) ON DELETE CASCADE,\n\tCONSTRAINT UNIQUE KEY `unique_user_group_id` (user_id, customer_key)\n) ENGINE=InnoDB DEFAULT CHARSET=latin1",
  "tableChanges" : [ ]
}

Schema Change Topic

As the schema history topic is for the internal use of debezium, it provides a schema change topic where external consumers can consume the schema change events. The topic name can be configured with topic.prefix (earlier database.server.name)

Snapshots

Debezium stores the snapshots of the database to provide high fault tolerance. In order to perform a snapshot, the connector first tries to get the global read lock that blocks the writes by the other clients and then reads the schema of all the tables and releases the lock. Acquiring a lock is very important because it helps in maintaining consistency as it blocks writes during that period. In case the global read lock is not possible, then it acquires table-level locks. More about it here.

There are different modes of snapshots. It can be configured with snapshot.mode . The most used modes are

initial used when you need the schema changes and the row level changes from the beginning. Schema changes are written to schema history and schema change topics and the data changes are written to <topic.prefix>.<table_name> .

schema_only takes the snapshot of only schema. This is useful if you don't want the entire data of the tables instead you only need the data from the moment you deployed. This mode is used if your tables contain dynamic data in an OLTP system.

when_needed takes the snapshot whenever it's necessary i.e. when the binlogs are deleted or the schema history topic is deleted etc...

Troubleshooting Debezium

The MySQL server is not configured to use a ROW binlog_format, which is required for this connector to work properly

io.debezium.DebeziumException: The MySQL server is not
configured to use a ROW binlog_format, which is required for this
connector to work properly. Change the MySQL configuration to use a
binlog_format=ROW and restart the connector.    at
io.debezium.connector.mysql.MySqlConnectorTask.validateBinlogConfiguration(MySqlConnectorTask.java:262) at
io.debezium.connector.mysql.MySqlConnectorTask.start(MySqlConnectorTask.java:86) at
io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:130)  at
org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232)    at

Change the MySQL configuration either through /etc/mysql.cnf or by executing the below command in the MySQL shell.

SET GLOBAL binlog_format = 'ROW';

Refer - MySQL Documentation

Database history topic retention.ms should be ‘-1’ or greater than 5 years

2022-01-06 13:08:55,904 WARN   ||  Database history topic
'dbhistory.dev-jigyasa-licious' option 'retention.ms' should be '-1' or greater than '157680000000' (5 years)
but is '604800000'   [io.debezium.relational.history.KafkaDatabaseHistory]

Set the retention.ms property of database history topic to -1 via kafka CLI.

The connector is trying to read binlog but this is no longer available on the server.

2022-01-06 13:08:56,405 ERROR  || 
WorkerSourceTask{id=jigyasa-mysql-connector-licious-2-0} Task threw an uncaught and unrecoverable exception.
Task is being killed and will not recover until manually restarted  
[org.apache.kafka.connect.runtime.WorkerTask] io.debezium.DebeziumException:
The connector is trying to read binlog starting at SourceInfo [currentGtid=null,
currentBinlogFilename=mysql-bin-changelog.032256, currentBinlogPosition=53270, currentRowNumber=0,
serverId=0, sourceTime=null, threadId=-1, currentQuery=null, tableIds=[], databaseName=null],
but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed.  


2022-01-06 13:08:56,405 INFO   ||  Connector requires binlog file
'mysql-bin-changelog.032256', but MySQL only has mysql-bin-changelog.032918, mysql-bin-changelog.032919,
mysql-bin-changelog.032920   [io.debezium.connector.mysql.MySqlConnectorTask]

Extend the binlog retention time using the commands mentioned above.
Delete the connector and re-deploy it again using REST API.

Could not find the first log file name in the binary log index file Error code: 1236; SQLSTATE: HY000.

org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
    at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42)
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1217)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:980)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:599)
    at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:857)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: io.debezium.DebeziumException: Could not find first log file name in binary log index file Error code: 1236; SQLSTATE: HY000.
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1172)
    ... 5 more
Caused by: com.github.shyiko.mysql.binlog.network.ServerException: Could not find first log file name in binary log index file
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:944)
    ... 3 more

This generally happens if you had upgraded MySQL, and cleaned up the binlogs. Delete the connector and re-deploy it again using REST API.

io.debezium.DebeziumException: Failed to read next byte from position 1640249837

org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
    at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42)
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1217)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:980)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:599)
    at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:857)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: io.debezium.DebeziumException: Failed to read next byte from position 1640249837
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1172)
    ... 5 more
Caused by: java.io.EOFException: Failed to read next byte from position 1640249837
    at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.read(ByteArrayInputStream.java:213)
    at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readInteger(ByteArrayInputStream.java:52)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:35)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:27)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:221)
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$1.nextEvent(MySqlStreamingChangeEventSource.java:230)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:952)
    ... 3 more

This happens if the binlog contains any special character that the connector is not able to process. One way to resolve this is to set event.processing.failure.handling.mode to warn in the connector configuration.

MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction

org.apache.kafka.connect.errors.ConnectException:An exception occurred in the change event producer. This connector will be
stopped.
    at
io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42)
    at
io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:115)
    at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by:
io.debezium.DebeziumException: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Lock wait
timeout exceeded; try restarting transaction
    at
io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:85)
    at
io.debezium.pipeline.ChangeEventSourceCoordinator.doSnapshot(ChangeEventSourceCoordinator.java:153)
    at

This happens when the debezium connector failed to get the global read lock or table level lock.

Increase the snapshot.lock.timeout.ms to 10-15 minutes and re-deploy the connector so that the debezium connector waits for some more time to acquire the lock.
If the above solution did not help, it means that the database is busy. Simply, kill all those clients and re-deploy the connector.

show open tables where in_use>0;
show full processlist;
kill <pid>;

Rate Limiting - Different Approaches and Implementation

Lokesh Sanapalli — Sun, 05 Feb 2023 05:59:23 +0000

Introduction

In software engineering, rate limiting has so many use cases. A few examples are

Limiting the number of times OTP is sent to mobile numbers.
Protecting the systems from DDOS (distributed denial of service) attacks.
Enterprise SaaS has pricing based on the number of requests per minute - so it is directly related to the business.

In this blog post, we will discuss different approaches of rate limiting.

First Principle Approach

Let's start with very basic first-principle thinking. What is rate limiting? let's break it down. Rate is the velocity at which an object is moving. We measure the velocity of the car in km per hour. Similarly, in computer science, the rate at which the requests are entered into a system is measured in requests per second. Limiting is self-explanatory. So, the problem statement is to control the rate at which the requests are coming to a system.

To measure the rate, we need to store the data somewhere. What are the options we have? hard disk and main memory. The disks are slow — This is not always true! it depends on how we use it! sequential reads and writes are relatively faster in the disk and we can further optimize it with the operating system's page caching into main memory — so the only option we have is main memory.

There are two popular libraries that store data in the main memory with high-level abstractions. They are Memcached and Redis. We need to leverage them to implement rate limiting. Let's see how we can do that.

Token Bucket Algorithm

The idea is simple, we will store the rate limit "X" at a cache key with an expiry of "T" seconds. The cache key can be an API key or a combination of user id, visitor cookie, IP address, etc... with the current minute as a suffix. For example, "u_lokesh1729:04" where the "u_lokesh1729" is the username of the user and "04" is the current minute.

Get the rate limit of a given cache key.
If the key is not found
1. set the key with "T" seconds as TTL
2. set the value as "X"
If the key is found
1. If the limit has been breached, reject the request.
2. If not, allow the request

import os
import redis

RATE_LIMIT = os.env.get("RATE_LIMIT", 50)

def rate_limit(key, rate):
    res = redis.get(key)
    if res is not None and res > RATE_LIMIT:
        raise TooManyRequestsException("rate limit breached")
    redis.incr(key)
    redis.expiry(key, 60)

Modified Token Bucket Algorithm

There is a slightly different approach for the same algorithm. Instead of adding the current minute to the key, we store two cache keys. One is mapping between the API key and the number of tokens. Another is a mapping between the API key and the last filled timestamp of the bucket in the epoch.

An example would be, "u_lokesh1729 -> 100" and "u_lokesh1729 -> 1674378362".

How do we determine the breach?

Get the count and the created timestamp.
If the count is greater than the limit, reject the request.
Else, find the diff between the current timestamp and the last filled timestamp of the bucket.
If the diff is greater than "T" seconds meaning the last window had already crossed.
1. Reset the rate limit and timestamp. There is no need to set the expiry. These are persistent keys.
Else
1. Compute and set the remaining tokens. How to compute? Say, the rate limit is "x" req/minute, and the diff between the bucket timestamp and the current timestamp is "y" seconds. The remaining tokens would be (y/1000) * x

Sliding Window Approach

Burst at edges

The downside of the above approach is that it is missing an edge case. The corners of adjacent windows can breach the rate limit but the algorithm won't be able to identify that. Refer to the below picture.

No Atomicity

Redis provides MULTI the command which wraps multiple redis commands and executes them as a single atomic block. So, all of them succeed or fail not partially. But, if we look at the above approaches, they cannot be fit in an atomic block because there is a business logic (checking the timestamp, tokens, etc...) happening outside. It can be solved with LUA scripting but that would complicate things.

Redis sorted sets

Redis provides a data structure called sorted sets. It is nothing but an array of tuples sorted by the score. An example would be

{
    "key": [
        (10, "abcd"),
        (11, "def"),
        (15, "xyz")
    ]
}

The key is the redis key at which the sorted set is stored. The TTL is set at the key level. The first item in the tuple is the score by which the list is stored. The second item in the tuple is the member associated with the score.

So, how do we use sorted sets to implement rate limiting? simple, we will use the epoch value of the timestamp as the score.

Remove the older entries which pass the current time using ZREMRANGEBYSCORE key -inf <curr_timestamp - 1 minute> .
Add the current timestamp to the set using ZADD key <timestamp> <timestamp> . We use the timestamp both as the score and the member.
Set the expiry of the key to one minute (or T seconds) using EXPIRE key 60
Get the count of the current values in the sorted set using ZCARD key

Using the count we got above, we can determine whether to allow or reject the request. Also, all the above commands are to be run in a single MULTI command.

What if the application is using Memcached? we can implement the above approach in the application code like this.

import cache
import time

def rate_limit(key, ttl, rate_limit):
    items = cache.get(key, [])
    time = int(time.time())
    # delete the items occurred one interval ago
    while items and items[-1] < time:
        items.pop()
    # add the current timestamp
    items.append(time)
    if len(items) > rate_limit:
        raise TooManyRequestsException("rate limit breached")
    # store the items with ttl
    cache.set(key, items, ttl)
    return True

Achieving high parallelism with kafka

Lokesh Sanapalli — Sun, 05 Feb 2023 05:51:39 +0000

In this post, we'll discuss consumer groups and how to achieve parallelism using them in Kafka. If you haven't read my previous blog posts about kafka, you will find them in the end.

Consumer Groups

As the name says, a consumer group is nothing but a group of consumers. Each consumer group is given a name. You can create as many consumer groups as you desire to listen to the same topic. Each consumer group can be at different offsets. That's the beauty of kafka. Here's how the consumer group works.

When you start a consumer, you give a name to it.
The consumer sends a request to the kafka broker by sending the consumer's name, topic, and partition number.
The kafka broker identifies the consumer by looking at its __consumer_offsets topic and returns the offset.
The consumer gets the offset and sends another request to get the payload and continues with the processing of the message.

Relationship between consumer group and partitions

There is always either a one-to-one or one-to-many relationship between the consumers in a consumer group and partition. Meaning, one consumer can read one partition or one consumer can read many partitions. But, multiple consumers belonging to a consumer group cannot read the same partition. Can you guess why? It is a design decision taken by the kafka engineering team to implement message delivery guarantee semantics. There are 3 delivery guarantees any message queue system can provide.

At least once - has the possibility of duplicate events to be consumed/produced
At most once - has the possibility of losing events
Exactly once - the ideal scenario

Assume that kafka allows one partition to be consumed by many consumers in a consumer group. In order to deliver "exactly once" semantics it needs to maintain the state of each message at multiple consumer levels. This becomes convoluted and makes the system complex. To avoid that, the kafka team had taken the decision to restrict one partition to be consumed by only one consumer in a consumer group. This way, the kafka broker needs to maintain a single offset (integer) for a partition.

Achieving High Parallelism

Parallelism is the ability to do multiple tasks at once. High parallelism guarantees high consistency in high throughput systems. Let's take a practical example where parallelism is very important. Consider an aggregator service in an e-commerce system consuming events from different services. If the aggregator service is consuming events at a slower pace, it results in a huge lag in the consumers thus serving inconsistent information to the end-users. So, it is important to have high parallelism in the system to maintain high consistency.

How do we achieve it in kafka? The number of partitions in a kafka topic is directly proportional to the parallelism. Based on the use case, we need to define the number of partitions at the beginning. Though it is possible to modify the partitions later, it takes downtime to do that because kafka needs to rebalance the data among partitions and replicas. Also, it is important not to create too many partitions. We can calculate the partitions required based on the throughput of messages.

Additional Read

https://lokesh1729.com/posts/kafka-internals-learn-kafka-in-depth

https://lokesh1729.com/posts/kafka-internals-learn-kafka-in-depth-part-2

OpenAPI - API documentation standard - Boon for the software engineers

Lokesh Sanapalli — Sun, 05 Feb 2023 05:46:01 +0000

OpenAPI - API documentation standard - Boon for the software engineers

Introduction

Your product manager brought a new product requirement. Frontend engineers and you (backend engineer) finalized a solution. They asked you for an API contract. You created a google doc (or confluence) and gave it to them. The contract changed later and you update the code as well as the document. Wouldn’t it be so nice if the documentation also lives in your code repository and automatically updates along with your code changes? That’s where OpenAPI comes in.

What is OpenAPI?

OpenAPI is a standard for defining the documentation of backend APIs. It is in the form of a file called a specification document. It can be in YAML or JSON format. OpenAPI sets standard syntax to define an API contract. You can read more about it here.

An example of an OpenAPI spec looks like this. A more detailed explanation of each field can be found here.

openapi: 3.1.0
info:
  title: Order Management System
  description: |
    Order Management provide APIs to create, cancel, deliver and reject orders
  version: 1.0.0
paths:
  /api/v1/order:
    post:
      summary: Creates a new order
      description: creates a new order
      paramters:
        ....
      responses:
        ....

OpenAPI is a Magic

You are maintaining a legacy system and you don’t have OpenAPI spec for it. Writing it from scratch manually would take a lot of time. Instead, you can use OpenAPI generators to generate specs automatically. You simply need to pass the response object to it. You can find them here - https://openapi.tools/#converters
The converters can also take HAR (HTTP Archive Format) file and generate a spec. You can get the HAR from chrome or firefox dev tools. Go to the network tab -> right-click on a request -> Save as HAR. List of such generators can be found here - https://openapi.tools/#auto-generators
You can generate boilerplate code stubs with OpenAPI spec. Yes! you heard it right! you only need to write the business logic that’s all, everything will be generated for you. There are so many generators available for every programming language. Find it here https://openapi-generator.tech. GitHub repo for the same https://github.com/OpenAPITools/openapi-generator & https://github.com/OpenAPITools/openapi-generator-cli

With the above tools, writing an API is very easy in 2 steps. Create a sample JSON response structure of your API. Generate OpenAPI spec and pass it to the generators to generate stubs.

OpenAPI UI Tools

There are two popular UI tools generated using OpenAPI spec. One is a widely used and popular swagger. Another is Redoc. The integration of these tools with any programming language is very easy. The guides for them are present on their respective websites.

How to generate OpenAPI schema

Existing APIs

If there are existing APIs in production, download HAR (from chrome/firefox devtools or use mitmproxy) and use HAR to OpenAPI tools to generate schema.

New APIs

Create a mock API using mocky
Hit the request from swagger hub
In the history, select the request and click on “CREATE API DEFINITION”
This will create OpenAPI spec but does not include response.
To include a response, first create JSON schema from your JSON using easy json schema.
Copy the JSON schema and paste it here

Credits to these StackOverflow answers

https://stackoverflow.com/a/49277426/5123867

https://stackoverflow.com/a/59738617/5123867

Additional Referrences

OpenAPI Homepage - https://www.openapis.org

OpenAPI Spec - https://spec.openapis.org/oas/v3.1.0

OpenAPI Documentation - https://oai.github.io/Documentation

OpenAPI Generators - https://openapi-generator.tech

OpenAPI Tools - https://openapi.tools