Thanks for subscribing! Look <strong>out</strong> for your first newsletter in your inbox soon! ... By entering your email address you agree to our Terms of Use and Privacy Policy and consent to receive emails from <strong>Time</strong> <strong>Out</strong> about news, events, offers and partner promotions.

World Travel Awards: Braga Named As Europe's Next Emerging Destination

This new restaurant in Tribeca is only open three days a week

This much-loved vegan café in east London needs your help to survive

Get the most recent info and news about Time Out Group on HackerNoon, where 10k+ technologists publish stories for 4M+ monthly readers.

Time Out Group plc, together with its subsidiaries, engages in media and entertainment business. It operates through two segments, Time Out Market and Time Out Media. The Time Out Market segment operates bars; and engages in retail, events, and sponsorship business activities. The Time Out Media segment engages in the sale of digital and print advertising, local marketing solutions, and live events tickets, as well as is involved in the e-commerce transactions and franchise activities. As of June 30, 2022, it operated in 333 cities in 59 countries. Time Out Group plc was founded in 1968 and is headquartered in London, the United Kingdom.

Time Out Group

Right after my first try of Kamal (MRSK) in the spring of 2023, I understood that an ideal use case would be running it as a GitHub Action. Almost a year passed, and my 30-line action has grown and become full-featured, configurable, and reusable. In this post, I will share the evolution of the action and the lessons learned. Before we start, let me remind you what Kamal is. Kamal is a library created by 37signals to orchestrate the deployment of Docker containers. Before switching to Kamal, I had a bunch of scripts and technologies to deploy my applications. Kamal allowed me to simplify the deployment process and make it more reliable. Also, there were already some GitHub workflows for running lints and tests. Ruby In this article, I will focus on deployment. Content Overview First try Action’s File Structure Pre Build Action Build & Deploy Action Workflow Definitions Conclusion First Try In an older article about Kamal, I posted the first version of the . It was a simple action that used the action to install Ruby, then to configure the SSH agent, prepared AWS credentials and then run the and command. GitHub Action ruby/setup-ruby webfactory/ssh-agent kamal envify kamal deploy The code is below. name: Kamal

on:
  push:
    branches:
      - main

jobs:
  spec:
    uses: ./.github/workflows/specs.yml
  lint:
    uses: ./.github/workflows/lint_code.yml

  build_and_deploy:
    needs: [spec, lint]
    runs-on: ubuntu-latest
    timeout-minutes: 20
    outputs:
      image: ${{ steps.build.outputs.image }}
    env:
      RAILS_ENV: production
    steps:
      - uses: actions/checkout@v3
        with:
          ref: ${{ github.event.workflow_run.head_branch }}

      - uses: webfactory/ssh-agent@v0.7.0
        with:
          ssh-private-key: ${{ secrets.SSH_PRIVATE_KEY }}

      - uses: ruby/setup-ruby@v1
        env:
          BUNDLE_GEMFILE: ./kamal/Gemfile
        with:
          ruby-version: 3.2.2
          bundler-cache: true

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2
        with:
          driver-opts: image=moby/buildkit:master

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-access-key-id    : ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region           : us-east-1
          mask-aws-account-id  : 'true'

      - name: Login to Amazon ECR
        id  : login-ecr
        uses: aws-actions/amazon-ecr-login@v1

      - name: Kamal Envify
        id  : kamal-envify
        env :
          KAMAL_REGISTRY_PASSWORD: ${{ steps.login-ecr.outputs.docker_password_YOUR_AWS_ACCOUNT_ID_dkr_ecr_YOUR_AWS_REGION_amazonaws_com }}
          DATABASE_URL: ${{ secrets.DATABASE_URL }}
          REDIS_URL: ${{ secrets.REDIS_URL }}
          RAILS_MASTER_KEY: ${{ secrets.RAILS_MASTER_KEY }}
          DOCKER_BUILDKIT: 1
          BUNDLE_GEMFILE: ./kamal/Gemfile
        run: |
          ./bin/kamal envify

      - name: Kamal Deploy
        id: kamal-deploy
        run: |
          ./bin/kamal deploy There are several things to pay attention to. First, at lines , I set up the  SSH agent to be able to connect to the instances and run Docker commands. The SSH private key is stored in the GitHub secrets, which is a convenient way to store sensitive data. 27-29 On lines I set up Docker Buildx, and it was very important to download Buildkit from the repository branch. At the moment of spring-winter of 2023, Buildkit didn’t support cache manifests for AWS ECR, and I had to use the branch to get the feature ( ). 38-41, master master Issue #876 On lines , I configured AWS credentials and logged in to Amazon ECR. It is not the most secure way to login to AWS, but was the easiest way to do it at that time. 43-53 This setup had several drawbacks. The first and the most important, the action was not reusable. To configure deployment to another environment, I had to copy-paste the action completely. Besides that, the action didn’t provide the ability to run deploy without running specs and lints. It was not a big deal, but sometimes I wanted to run deploy separately. The action didn’t provide the ability to run deploy to another branch. Sometimes I wanted to restart the Traefik container. This could be done from the local machine, but for me, it would be better to do it from the action. Action’s File Structure We incrementally improved the action, added new features, and made it more configurable. Finally, we come to the set of actions that can be used in almost any project. For a better understanding of the changes, let’s look at the directory structure of the actions first. .github/
├─ workflows/
   ├─ build_deploy/
      ├─ action.yaml
   ├─ pre-build/
      ├─ action.yaml
   ├─ 01.build_deploy_production.yaml
   ├─ 02.build_deploy_staging.yaml
   ├─ 03.database_backup.yaml
   ├─ 04.build_deploy_manually.yaml
   ├─ 05.validate_pull_request.yaml
   ├─ 06.kamal_run_command.yaml
   ├─ _lint.yaml
   ├─ _specs.yaml You may notice two directories, and , both have file inside. These are that include all the necessary steps to build and deploy the application. Also, some workflows are named with a leading underscore. These are reusable workflows that consist of several jobs and steps. build_deploy pre-deploy action.yaml composite actions How are Composite Actions different from Reusable Workflows? Composite Actions allow you to bundle multiple existing workflow steps into a single action. A Composite Action cannot be used without a repo checkout while Reusable Workflows can be used without a checkout. A Reusable Workflow can include multiple jobs and multiple steps within those jobs. However, Composite Actions can only have one job. Reusable Workflow can use Secrets by declaring them to a workflow via parameters while Composite Actions cannot use Secrets in a flexible way. Files with numeric prefixes are main workflow definitions that call the reusable workflows and composite actions. They cover the most common usecases for a modern Rails project: deploy to production and staging, database backup, manual deploy, pull request validation, and Kamal command execution. Pre Build Action Let’s start with the action. It is a composite action that includes all the necessary steps to prepare the environment for the build and deploy action. pre-build The file is below. # pre-build/action.yml

name: Pre-Build

inputs:
  database-url:
    type: string
  redis-url:
    type: string
  rails-master-key:
    type: string
  aws_role_access:
    type: string
  ssh_private_key:
    type: string
  environment:
    type: string

runs:
  using: composite
  steps:
    - uses: webfactory/ssh-agent@v0.8.0
      with:
        ssh-private-key: ${{ inputs.ssh-private-key }}

    - uses: ruby/setup-ruby@v1
      env:
        BUNDLE_GEMFILE: ./Gemfile
      with:
        ruby-version: .ruby-version
        bundler-cache: true

    - name: Set up Docker Buildx
      uses: docker/setup-buildx-action@v3

    - name: aws-cred-configure
      uses: aws-actions/configure-aws-credentials@v4
      with:
        role-to-assume   : ${{ inputs.aws-role-access }}
        role-session-name: samplerolesession
        aws-region       : es-east-1
        mask-aws-account-id: 'true'

    - name: login-to-aws-ecr
      id: login-ecr
      uses: aws-actions/amazon-ecr-login@v2
      with:
        mask-password: 'true'

    - name: Kamal Envify
      shell: bash
      id  : kamal-envify
      env :
        KAMAL_REGISTRY_PASSWORD: ${{ steps.login-ecr.outputs.docker_password_AWS_ACCOUNT_ID_dkr_ecr_eu_west_2_amazonaws_com }}
        DATABASE_URL: ${{ inputs.database-url }}
        REDIS_URL: ${{ inputs.redis-url }}
        RAILS_MASTER_KEY: ${{ inputs.rails-master-key }}
        DOCKER_BUILDKIT: 1
      run: |
        ./bin/kamal envify --destination=${{ inputs.environment }} Let’s walk line by line through the file. Lines describe the inputs of the action and are not interesting to us. The first step is to set up an SSH agent, and it is the same as in the first version of the action. However, as you may see, it uses from the action inputs, which allows us to use the action in different environments. After that, I setup Ruby, and it is also the same as in the first version of the action. 5-17 ssh-private-key The Docker Buildx setup step differs from the initial workflow definition. It has been updated to version 3 and now out the box, so we don’t need to define anymore. supports AWS ECR image cache driver-opts The next step is to configure AWS credentials. It is completely different from what I initially had. Instead of and authentication, I switched to the UIDC role-based authentication, which is more secure and GitHub . If you need a more detailed explanation of how to configure OpenID Connect in AWS, I suggest you read this . After the authentication is done, I login to Amazon ECR on lines . access-key-id secret-access-key advises to use it excellent guide 44-48 The final step is to run , which prepares environment variables for the deployment. The command is the same as in the first version of the action, but I added the flag to the command, which allows me to deploy to different environments. kamal-envify --destination Build & Deploy Action The next composite action is defined in folder and it is relatively simple. build_deploy # build-deploy/action.yml
name: Build & Deploy

inputs:
  environment:
    type: string

runs:
  using: composite
  steps:
    - name: Kamal Deploy
      shell: bash
      id: kamal-deploy
      run: |
        ./bin/kamal deploy --destination=${{ inputs.environment }}

    - name: Kamal Release
      shell: bash
      if: ${{ cancelled() }}
      run: |
        ./bin/kamal lock release --destination=${{ inputs.environment }} Since all preparations are done in , when this action starts, I am ready to run the command with the selected environment. pre-build kamal deploy Kamal creates a lock file before starting the build and deployment process. Usually, lock is released when deployment is finished, but if the deployment is canceled, the lock is not released. And the next workflow run will fail. To avoid this, I added the command to the action. Later this article, we will use this small hack to handle concurrent deployments correctly. kamal lock release Workflow Definitions The main workflow definitions are very simple. They just call the composite action together with reusable workflows and pass the necessary parameters. Below is an example of the file. 01.build_deploy_production.yaml # 01.build_deploy_production.yaml
name: 01. Build & Deploy Production

permissions:
  id-token: write
  contents: read

on:
  release:
    types: [published]

jobs:
  spec:
    uses: ./.github/workflows/_specs.yaml
    secrets: inherit

  lint:
    uses: ./.github/workflows/_lint_code.yaml
    secrets: inherit

  build_and_deploy:
    name: build-deploy-production
    concurrency:
      group: production_environment
      cancel-in-progress: true
    environment:
      name: production
      url: https://onetribe.team
    needs:
      - spec
      - lint
    runs-on: ubuntu-latest
    timeout-minutes: 20
    steps:
      - uses: actions/checkout@v4
        with:
          ref: ${{ github.event.workflow_run.head_branch }}

      - name: Pre Build
        uses: ./.github/workflows/pre-build
        with:
          database-url: ${{ secrets.DATABASE_URL_PRODUCTION_ADMIN }}
          redis-url: ${{ secrets.REDIS_URL_PRODUCTION_ADMIN }}
          rails-master-key: ${{ secrets.RAILS_MASTER_KEY }}
          aws-role-access: ${{ secrets.AWS_ROLE_ACCESS }}
          ssh-private-key: ${{ secrets.SSH_PRIVATE_KEY }}
          environment: production

      - name: Build & Deploy
        uses: ./.github/workflows/build-deploy
        with:
          environment: production The section in lines deserves attention in the file. It allows me to run only one deployment at a time. GitHub has a great that covers all possible use cases. If a deployment is already running, it will be canceled, and the step from the previous workflow run will be executed. This is an essential feature because it allows me to avoid concurrent deployments and handle them correctly. concurrency 23-25 documentation section "Kamal Release" Staging deploy is defined in the file and is similar to production, except for the event that starts the workflow: for staging deploy I use the event to GIT branch, instead of the . 02.build_deploy_staging.yaml push staging release # 02.build_deploy_staging.yaml
name: 02. Build Staging

permissions:
  id-token: write
  contents: read

on:
  push:
    branches:
      - staging

# ... In this article, I will not cover database backup workflow, as defined in the file, because it is not related to the theme of the article. However, let’s look at , , and files. 03.database_backup.yaml 04.deploy_manually.yaml 05.validate_pull_request.yaml 06.kamal_run_command.yaml The file is below. 04.build_deploy_manually.yaml # 04.build_deploy_manually.yaml
name: 04. Deploy Manually

permissions:
  id-token: write
  contents: read

on:
  workflow_dispatch:
    inputs:
      environment:
        description: 'Environment'
        required: true
        default: 'staging'
        type: choice
        options:
          - production
          - staging

jobs:
  build-production:
    name: deploy-production
    concurrency:
      group: production_environment
      cancel-in-progress: true
    environment:
      name: production
      url: https://onetribe.team
    if: ${{ github.event.inputs.environment == 'production' }}
    runs-on: ubuntu-latest
    timeout-minutes: 20
    steps:
      - uses: actions/checkout@v4
        with:
          ref: ${{ github.event.workflow_run.head_branch }}

      - name: Pre Build
        uses: ./.github/workflows/pre-build
        with:
          database-url: ${{ secrets.DATABASE_URL_PRODUCTION_ADMIN }}
          redis-url: ${{ secrets.REDIS_URL_PRODUCTION_ADMIN }}
          rails-master-key: ${{ secrets.RAILS_MASTER_KEY }}
          aws-role-access: ${{ secrets.AWS_ROLE_ACCESS }}
          ssh-private-key: ${{ secrets.SSH_PRIVATE_KEY }}
          environment: production

      - name: Build and Deploy
        uses: ./.github/workflows/build-deploy
        with:
          environment: production

  build-staging:
    name: deploy-staging
    concurrency:
      group: staging_environment
      cancel-in-progress: true

    # ...
    # staging deploy is similar to production, described above and I will not show it completely. Pull request validation is defined in the , it is the most small and simple workflow. 05.validate_pull_request.yaml It runs specs and lints, triggered by the event and also can be triggered manually. pull_request # 05.validate_pull_request.yaml
name: 05. Validate Pull Request

permissions:
  id-token: write
  contents: read

on:
  pull_request:
  workflow_dispatch:

jobs:
  spec:
    uses: ./.github/workflows/_specs.yaml
    secrets: inherit

  lint:
    uses: ./.github/workflows/_lint_code.yaml
    secrets: inherit The last workflow that I want to cover in this article is the file. Sometimes I need to restart the Traefik container or maybe start or stop accessory containers. I can do it from the local machine, but this requires environment setup and is not always 06.kamal_run_command.yaml convenient. This workflow allows me to run any command from the list of predefined commands. name: 06. Kamal run command

permissions:
  id-token: write
  contents: read

on:
  workflow_dispatch:
    inputs:
      command:
        description: 'Commands'
        required: true
        type: choice
        options:
          - traefik reboot --rolling
          - accessory reboot pg_hero
      environment:
        description: 'Environment'
        required: true
        type: choice
        options:
          - staging
          - production

jobs:
  kamal_run_command:
    name: Kamal run command
    runs-on: ubuntu-latest
    timeout-minutes: 20
    concurrency:
      group: ${{ github.event.inputs.environment }}_environment
      cancel-in-progress: false
    environment:
      name: ${{ github.event.inputs.environment }}
    steps:
      - uses: actions/checkout@v4
        with:
          ref: ${{ github.event.workflow_run.head_branch }}

      - uses: ./.github/workflows/pre-build
        name: Pre Build
        with:
          database-url: ${{ github.event.inputs.environment == 'production' && secrets.DATABASE_URL_PRODUCTION || secrets.DATABASE_URL_STAGING }}
          redis-url: ${{ github.event.inputs.environment == 'production' && secrets.REDIS_URL_PRODUCTION || secrets.REDIS_URL_STAGING }}
          rails-master-key: ${{ secrets.RAILS_MASTER_KEY }}
          aws-role-access: ${{ secrets.AWS_ROLE_ACCESS }}
          ssh-private-key: ${{ secrets.SSH_PRIVATE_KEY }}
          environment: ${{ github.event.inputs.environment }}

      - name: kamal ${{ github.event.inputs.command }} --destination=${{ github.event.inputs.environment }}
        run: |
          ./bin/kamal ${{ github.event.inputs.command }} --destination=${{ github.event.inputs.environment }} Conclusion The action has grown from a simple 30-line action to a set of reusable workflows and composite actions. It is now full-featured, configurable, and reusable. It allows me to run deploy to different environments, run deploy without running specs and lints, and restart Traefik containers and accessories. I have used this or a similar setup of workflows for about six months and what can I say? It covers all my needs and can be easily adapted for any new features. I hope this article will help you to build your action and workflows. If you have any questions, feel free to ask me in the comments. Also published . here

Building a Reusable Deployment Workflow with Kamal and GitHub Actions

In this post, we’ll , the consequences for the application, as well as steps to minimize downtimes in event processing when consumer group members change. learn how ’ Consumers behave differently from regular Kafka Consumers Kafka Streams With the , a containerised when one app instance (group member) is removed or restarted. For real-time data streaming workloads with a low e2e latency as an NFR (non-functional requirement), such a long ‘rebalance downtime’ often is unacceptable. default configuration stateless Streams app pauses processing for >45s Fortunately, there’s a simple yet efficient solution to address this problem. As a bonus, we will look under the hood of Kafka Consumer Groups, the Group Coordinator & Rebalance Protocol, and measure, analyze, and evaluate a simulation of a group member (replica) re-creation running on Kubernetes. ... ? here's a : TLDR spoiler props.put("internal.leave.group.on.close", true); Content Overview Theory Example Scenario: Kubernetes Pod Evicted… and Replaced Wait, 48s? Really??? Re-do the Example with ‘leaveGroupOnClose’ Pro Tips Conclusion Footnotes References and Further Reading Theory Regular Consumer Behaviour Let’s briefly recap on Kafka Consumers and Consumer groups. An Apache Kafka® is a client application that subscribes to (reads and processes) events. Consumer A is a set of consumers which cooperate to consume data from some topics. The partitions of all the topics are divided among the consumers in the group. As new group members arrive and old members leave, the partitions are re-assigned so that each member receives a proportional share of the partitions. This is known as the group. consumer group rebalancing (…) One of the brokers is designated as the group’s and is responsible for managing the members of the group as well as their partition assignments. coordinator (…) When the consumer starts up, it finds the coordinator for its group and sends a request to join the group. The coordinator then begins a group rebalance so that the new member is assigned its fair share of the group’s partitions. Every rebalance results in a new of the group. generation Each member in the group must send heartbeats to the coordinator in order to remain a member of the group. If no heartbeat is received before expiration of the configured , then the coordinator will kick the member out of the group and reassign its partitions to another member. session timeout (Source: ) Kafka Consumer | Confluent Documentation When a consumer leaves a group due to a controlled shutdown or a crash, its partitions are reassigned automatically to other consumers. Similarly, when a consumer (re) joins an existing group, all partitions are rebalanced among the group members. This dynamic group cooperation is facilitated by the . Kafka Rebalance Protocol For a rebalance scenario where one instance is stopped, the Consumer sends a request to the coordinator before stopping (as part of a graceful shutdown, ), which triggers a rebalance. LeaveGroup Consumer#close() During the entire rebalancing process, i.e. as long as the partitions are not reassigned, consumers no longer process any data. Fortunately, rebalancing is very fast, typically between anything from 50ms to seconds. It may vary depending on different factors, such as load on your Kafka cluster or the complexity of your Streams topology (no. of input topics, streams tasks := partitions, and state stores, … -> total no. of consumers). != Streams Consumer Behaviour For Kafka Streams, some config properties are overridden via ( ). One of those properties is , set to (enabled by default for Consumers). StreamsConfig.CONSUMER_DEFAULT_OVERRIDES "internal.leave.group.on.close" false regular Please note it’s a config, which may change without prior notice with new releases. Reference: . non-public ConsumerConfig.LEAVE_GROUP_ON_CLOSE_CONFIG This means Consumers will not send requests when stopped but will be removed by the coordinator only when the Consumer session times out (ref. ). The (note: was 10s before the Kafka 3.0.0 release, ref ). Consequently, no data is processed for more than 45 seconds for tasks assigned to the Consumer that had been stopped. LeaveGroup session.timeout.ms default Consumer session timeout is 45s KIP-735 It even worsens if a new Consumer (re) joins the group while suspected dead (no more heartbeats received), where all consumers shut down, and task assignment is blocked until the timeout is exceeded. The coordinator evicts the old Consumer that had been stopped from the group. Until then, processing comes completely to a halt for all tasks, also known as ‘stop-the-world’ rebalancing. While the ‘incremental cooperative rebalancing protocol’ introduced with Kafka 2.5 avoids ‘stop-the-world’ rebalancing for Consumers, the mentioned Kafka Streams overrides nullify some aspects. regular Example Scenario: Kubernetes Pod Evicted … and Replaced Running your apps on Kubernetes takes a longer route to achieve a robust, highly-available deployment. Kubernetes monitors your containers’ health, allows you to scale, and ensures all desired replicas are up and running according to your spec. But still, to be truly elastic and minimize downtime of your data stream processing, your application must be able to handle Pods (/container) to be restarted, evicted, and re-created gracefully. There are many potential causes, e.g. application upgrades ( ), k8s cluster security patching, (auto-)scaling, resource shortage, or k8s nodes running on Spot instances being interrupted. CI/CD Next, we look at a simple yet common example. Stateless Kafka Streams app, 6 streams tasks, running on Kubernetes as Deployment, with 3 replicas. Example infrastructure setup: One pod is terminated and successively replaced. Scenario: Initial state, all 3 pods are running & healthy, the streams app is processing, balanced task assignment Pod (P1.1) terminated (deleted) by k8s, shutting down gracefully A replacement Pod (P1.2) is scheduled & placed The final state, the replacement Pod is running & healthy, the streams app is processing balanced task assignment. I would like to share a screenshot depicting the consumer lag metrics, rendered in Grafana, for a simulation of our scenario. Let’s walk through the results and explain the behavior: the Pod (P1.1) is terminated and stops processing. The consumer lag of partitions starts to build up 18:34:00: [1,4] the replacement Pod (P1.2) has come up; the streams task sends a to the group coordinator 18:34:20: JoinGroup rebalancing triggered, assignments revoked, pauses - due to no heartbeats received from (P1.1) 18:34:21: all consumers pause processing, waiting for assignment; lag starts to build up for all partitions 18:34:21: rebalancing continues, new assignment, processing continues 18:34:45: all consumers caught up; consumer lags are back to healthy jitter 18:34:48: To better illustrate everything that is happening over time, here’s a time bar diagram highlighting all important steps: Here are the belonging application logs for the rebalancing, which occurred at and took . 16:34:44.988 92ms 2023-06-17 16:34:44,988 INFO State transition from RUNNING to REBALANCING
2023-06-17 16:34:45,080 INFO State transition from REBALANCING to RUNNING So, we can conclude the following downtimes: partitions : [2,5] 48s partitions : While the actual rebalancing took only . [0,1,2,3,4] 25s 92ms 😵 Wait, 48s? Really??? Depending on your stream processing use case, 45s+ downtime might be no big deal, but , it’s a massive . for real-time low-latency data streams breach of the NFR So let’s see what options we’ve got to mitigate: Option 1: Lower consumer session timeout Since the session timeout determines the downtime, one way to mitigate is to reduce . Don’t forget to decrease the value of to ensure three heartbeats plus a buffer can fit within the timeout period. session.timeout.ms heartbeat.interval.ms session.timeout.ms=6000
heartbeat.interval.ms=1500 Read the config here: Kafka Consumer Configurations Option 2: Enable ‘leaveGroupOnClose’ …but why work with timeouts when it’s perfectly valid to have your Streams Consumers notify the coordinator when closing down?!? stateless To enable ‘leaveGroupOnClose’ (overriding the 😜), configure your Kafka Streams app with the following property: override internal.leave.group.on.close=true Please note it’s a config, which may change without prior notice with new releases. Reference: . non-public ConsumerConfig.LEAVE_GROUP_ON_CLOSE_CONFIG Re-do the Example with ‘leaveGroupOnClose’ 🚀 Drum roll 🥁 … and here, without further ado, are the results: As we can ( ) see - the two rebalancings complete so fast that there's not even the slightest consumer lag increase visible in the metrics. Here’s the visual explanation: not Finally, here are also the application logs showing the timings of the rebalancing, which happened twice. One at that took , and the other at in . 17:46:00.332 92ms 17:46:21.361 98ms 2023-06-17 17:46:00,332 INFO State transition from RUNNING to REBALANCING
2023-06-17 17:46:00,424 INFO State transition from REBALANCING to RUNNING
2023-06-17 17:46:21,361 INFO State transition from RUNNING to REBALANCING
2023-06-17 17:46:21,458 INFO State transition from REBALANCING to RUNNING Pro Tips Stateless <> Stateful This post recommends setting for Kafka Streams applications. internal.leave.group.on.close=true stateless (!) Before implementing for stateful applications, it is crucial to understand all potential consequences. internal.leave.group.on.close=true Unfortunately, my evaluation using `internal.leave.group.on.close=true` in combination with standby replicas was not very promising. The expected fluent task re-assignment to hot standby while one replica "restarts" - and subsequent re-distribution of tasks, does not work. The Kafka Streams specific `HighAvailabilityTaskAssignor` has known issues such as uneven task assignment, frozen warmup tasks ('task movement'), and not recognising caught-up standby tasks when the consumer group changes. Please note there are plans to address those issues with the next version of the Consumer Rebalance Protocol (see [footnotes]( )). Often the best plan to keep downtimes low during rebalance for stateful apps is to stick with RocksDB + StatefulSet + PersistentVolumes + restart within (!) the session timeout => re-join with previous assignment, re-use RocksDB state, and avoid rebalancing entirely... #footnotes Alternatively, take a look at , introduced in an . kafka-streams-cassandra-state-store earlier blog post k8s Deployment .spec.minReadySeconds Frequently rebalancing within a short timeframe can cause consumer delays and strain the Kafka cluster. If your application/container has a quick restart time, such as when running as a GraalVM native executable, it’s worth considering the use of to maintain control and ensure upgrades occur in a controlled manner. This will help prevent frequent rebalancing within a short timeframe. .spec.minReadySeconds Conclusion By configuring your Kafka Streams app with , a graceful and tasks are re-assigned to other active members within the group. The while . As a result, your applications enables interruption-free CI/CD and can be auto-scaled. internal.leave.group.on.close=true shutdown immediately triggers a rebalancing process processing downtime is significantly reduced also improving elasticity and resilience Please note that this recommendation only applies to streams applications.! Tread carefully for stateful topologies, and do your homework!  Remember that stateless is a config, which may change without prior notice with new releases. Always check the source code for changes when upgrading the Kafka Streams dependency. internal.leave.group.on.close non-public Footnotes When writing this blog post, the latest version of kafka-streams was 3.4.1. There’s a ticket from June 2018 proposing to make the config public. The ticket is closed as . Concerns of the core developer team can be found in the discussion. KAFKA-6995 ’Won’t Fix’ Looking into the crystal ball: A Kafka Design Proposal (KIP) is in progress to introduce a new group membership and rebalance protocol for the Kafka Consumer and, by extensions, Kafka Streams.=> KIP-848: The Next Generation of the Consumer Rebalance Protocol It was also introduced on : Current 2022 The Next Generation of the Consumer Rebalance Protocol With David Jacot | UK The application + docker-compose setup that was put together for this article can be found on the thriving-dev GitHub Organisation::icon{name="mdi-github" class="inline -mt-0.5 w-6 h-6"} https://github.com/thriving-dev/kafka-streams-leave-group-on-close Many thanks to for proofreading the blog post! 🙇 @MatthiasJSax References and Further Reading Kafka Consumer Group Rebalance (1 of 2) Kafka Consumer Group Rebalance (2 of 2) Kafka-streams delay to kick rebalancing on consumer graceful shutdown - Stack Overflow Cooperative Rebalancing in the Kafka Consumer, Streams & ksqlDB Apache Kafka Rebalance Protocol, or the magic behind your streams applications | by Florian Hussonnois | StreamThoughts | Medium API that forces the member to leave the consumer group - Apache Kafka - Apache Software Foundation KIP-812: Introduce another form of the KafkaStreams.close() This post was originally published on . Thriving.dev

Minimizing Rebalance Downtime: Optimizing Stateless Kafka Streams Apps (x450)

In today’s digital landscape, creating a blog is just the first step in establishing an online presence. To attract readers and build a following, bloggers need to promote their content effectively. In this blog post, we’ll explore the topic of automated blog promotion and how to use ChatGPT, Twitter API, and AWS to create a blog promotion toolkit that can help bloggers promote their content more efficiently. The toolkit is designed to automate the promotion process using cutting-edge technologies, making it easier for bloggers to reach their target audience. By following the steps outlined in this post, you’ll be able to create your own automated blog promotion toolkit and take your blog promotion efforts to the next level. Let’s dive in and see how ChatGPT, Twitter API, and AWS can help you promote your blog like a pro! Requirements: . The toolkit must run automatically on a schedule, without requiring manual intervention, to save users time and effort in promoting their content. Schedule automation . The toolkit should be easy to extend, allowing users to add new promotion tools as needed. The MVP version of the toolkit will contain only one tool designed to engage new readers by replying to tweets using OpenAI APIs and ChatGPT, with AI text generation to ensure that the Twitter bot responds using relevant context. In the future, we may want to add functionality (publishing to relevant communities) or even expand to new platforms (e.g., LinkedIn). Extendibility . The tool must be polite and respectful to Twitter accounts, avoiding multiple replies to the same user within a short period to prevent appearing annoying or spammy. Politeness The toolkit should be able to run both in the AWS cloud and locally, providing users with the flexibility to choose the deployment method that works best for them. Hybrid runtime. Problem Creating high-quality content for a blog is a challenging task, but promoting that content to reach a wider audience can be even more difficult. Social media platforms like Twitter offer a powerful way to share blog posts and engage with readers, but manually managing a Twitter account and replying to tweets can be time-consuming and repetitive. ' Furthermore, engaging with Twitter users in a way that is polite and respectful can be challenging, particularly when trying to reach a large number of users. These challenges can make it difficult for bloggers to effectively promote their content on Twitter and reach new readers. In the next section, we will explore how our blog promotion toolkit can help address these challenges and enable bloggers to reach their promotion goals. Solution Our blog promotion tool aims to simplify the process of promoting blog content on Twitter by automating the process of finding relevant tweets and engaging with Twitter users. The tool takes a list of blog posts, each with its own set of hashtags and metadata. These hashtags and metadata are configured manually by the author of the blog (using the same ChatGPT usually 😀) and extracted directly from the HTML. Once the hashtags have been extracted, the toolkit passes them to the Twitter SearchTweets API, which returns a list of relevant tweets from oldest to newest. The toolkit then checks each tweet for various requirements, such as politeness and the author’s audience size. If a tweet qualifies, the toolkit feeds the tweet URL and blog post URL to the OpenAI API, which generates a relevant reply using the ChatGPT language model. Finally, the toolkit sends the generated reply using the Twitter Manage Tweets API. We can visualize this flow as a pipeline with the following key components: By automating the process of finding relevant tweets and generating polite and engaging responses, our blog promotion toolkit can help bloggers save time and effort in promoting their content on Twitter. Additionally, the ability to extend the toolkit with new promotion tools in the future can provide even more ways to engage with readers and promote blog content. Tools and Libraries Our blog promotion toolkit is built using a variety of tools and libraries, including: : A popular programming language that is widely used in data analysis, machine learning, and web development. This is my first time using this language to build something meaningful, so please don’t judge me harshly. Python : A Python library that provides easy access to the Twitter API, allowing developers to build applications that interact with Twitter. I will be using V2 . Tweepy Client : A Python library that provides access to OpenAI’s powerful language models, including GPT-3. OpenAI library : A Python library for web scraping and parsing HTML documents. BeautifulSoup : (for storage), (for computing), (for event scheduling to trigger Lambda), and, of course, my favorite to integrate all of these services together into a working SAM template and deploy it via . AWS services DynamoDB AWS Lambda AWS CloudWatch AWS App Composer AWS CloudFormation Implementation TLDR; Link to the Code GitHub repo: https://github.com/sr-shifu/blog-promo-toolkit Building Infrastructure As mentioned earlier, I will use AWS App Composer UI to build my SAM template. I will not go deep into the details (you can find more information about how to work with it in one of my previous ) — I will just leave my final diagram here: posts Infrastructure is pretty simple: lambda function. I decided to set its timeout to the maximum (15 mins) to give it as much time as possible: OpenAI API can be slow, and Twitter search API might have pretty long cooldown timeouts after being throttled. EngageTweets Event rule (AWS::Events::Rule) that will trigger lambda function using this cron expression (unfortunately, UI does not support this configuration, so you need to write it by hand): cron(0 18 * * ? *). This expression can be decoded to human language as . EngageTweets "at 18:00 UTC every day" Secrets Manager instance to store tokens to access Twitter and OpenAI APIs. lambda will consume them via environment variables (again, need to configure manually in the template). IntegrationTokens EngageTweets table. It will be used to store information about tweets that have been already replied — it would be used by both the and components. It will use userId as a partitioning key and tweetId as a range (sort) key. RepliedTweets SearchTweets TweetFilter There are also 2 additional attributes that are worth mentioning: : expiration key (UNIX time of when the record needs to be deleted from the table). The idea behind this is to make storage more efficient. Twitter’s SearchTweets can search tweets only for the last 7 days, so there is no need to store tweet reply data forever. We can reply to the tweet, set TTL to be current_time + 7d , and make sure that we will not disturb the author during the next 7 days or will not reply to the same tweet again. TTL API : query string that was used to search this tweet (about this later). The idea here is to build Global Secondary Index (GSI) using searchKey as the primary key and tweetId as a range key. Using this index, we can find the latest tweet that was replied to in case the previous function execution was interrupted (for instance, exceeded lambda timeout). You may ask why don't we store reply time in the table and use it as a sort key instead. The answer is simple: Twitter uses IDs that guarantee that all tweets follow the rough order ( ). Search API also supports since_id, a request parameter that will give us all tweets that were sent after the last replied tweet id stored in our table. In case our tool didn't run during the last 7 days and all data in the table was purged, it's not a problem at all - remember, API returns only the recent 7 days. searchKey Snowflake roughly sortable Deeper Into the Code Initialize Tweepy client: client = tweepy.Client(
    bearer_token=bearer_token,
    consumer_key = consumer_key,
    consumer_secret = consumer_secret,
    access_token = access_token,
    access_token_secret = access_token_secret,
    wait_on_rate_limit = True
) With , the client will swallow all throttling exceptions and wait for the API to cool down. wait_on_rate_limit = True Generate reply using openAI: def generate_tweet_reply_message(tweet_url, post_url, lang = 'en'):
    prompt = f"Reply to tweet {tweet_url}. Reply must include link to article {post_url} and engage to follow @TheSameTech{' using ' + lang + ' language' if lang != 'en' else ''}. Don't exceed {str(MAX_TWITTER_MESSAGE_LENGTH)} chars"
    completions = openai.Completion.create(
        engine="text-davinci-003",
        prompt=prompt,
        max_tokens=300,
        n=1,
        stop=None,
        temperature=0.7,
    )
    author_id = re.search(r"twitter\.com/([^/]+)/status", tweet_url).group(1)
    # this is one of the weird stuff I noticed - sometimes ChatGPT tags author using their ID, and not account name
    return completions.choices[0].text.replace("\n\n", "").replace(f" @{author_id}", "") Parameters: : This parameter specifies the ID of the OpenAI language model to use for generating text. In this example, the text-DaVinci-003 model (ChatGPT-3) is used, which is one of OpenAI's most advanced models. engine : This parameter specifies the text prompt to use as input to the language model. In the blog promotion toolkit, the prompt will be the relevant tweet that the ChatGPT algorithm will be generating a response to. prompt : This parameter specifies the maximum number of tokens (words or punctuation marks) that the language model should generate in its response. max_tokens : This parameter specifies the number of responses to generate. In this case, only one response will be generated. n : This parameter specifies a sequence of tokens that should be used as a stopping point for the language model's response. In this case, no stopping sequence is specified. stop : This parameter controls the randomness of the language model's responses. A higher temperature value will produce more creative and varied responses, while a lower temperature value will produce more predictable and conservative responses. In this example, a temperature of 0.7 is used, which should produce responses that are creative but not too unpredictable. temperature : OpenAI released a new gpt-3.5-turbo language model a few days after I wrote this post ( ). It's priced at $0.002 per 1K tokens, which is 10x cheaper than the existing GPT-3.5 models. UPDATE Mar 7, 2023 Search tweets: twitter_metadata = extract_twitter_metadata(post_url, tagsSelector='.post-tags')
        keywords, hash_tags, description, *rest = twitter_metadata
        combos = list(combinations(hash_tags, 2))
        for combo in(combos):
            hash_tags_string = " ".join(combo)
            latest_tweet_id = get_latest_activity(hash_tags_string)
            if latest_tweet_id is None and search_days_ago is not None:
                start_time=(datetime.datetime.now() - datetime.timedelta(days=search_days_ago)).strftime("%Y-%m-%dT%H:%M:%SZ")
            tweets = search_recent_tweets_with_pagination(query=hash_tags_string, max_results = 100, start_time=start_time, latest_tweet_id=latest_tweet_id, tweet_fields=['id', 'author_id', 'created_at', 'in_reply_to_user_id', 'lang'])
            # do other stuff (filtering, generating reply, promoting) After Twitter metadata is extracted from HTML, I use hashtags as search criteria to find relevant tweets. Usually, every blog post has about 4–5 tags associated with it. If I pass down all of them together, most likely Twitter API will return nothing — that’s why I combine them by pairs and use every pair as a search key (so every blog post will generate from 6 to 10 search requests in general). Running Locally With SAM, you don’t need to do a lot. Only execute 3 commands, and you are good to go. If you want to run the local version of DynamoDB, please follow the README instructions in my GitHub repo. sam build
sam local start-lambda
aws lambda invoke --function-name "EngageTweets" --endpoint-url "http://127.0.0.1:3001" --no-verify-ssl out.txt It’s the same simple using a Python executable: cd src/engage-tweets-lambda
pip install -r requirements.txt
source env/bin/activate   
TABLE_NAME=PromotedTweets python engage_tweets.py Just make sure you have all tokens stored in your local file. .env Deploying to AWS One command you need to know: sam deploy Limitations Twitter API rate limit: Twitter API has a rate limit that limits the number of API calls that can be made per user per 15-minute window. This means that if a user exceeds the rate limit, they may not be able to make any further API calls until the limit resets (but we already got this covered — see previous section). Twitter monthly limit to search tweets: Twitter limits the number of tweets that can be searched in a given month. The current limit is 500,000 tweets per month per developer account, which is also subject to change. OpenAI API is not free: The OpenAI API charges based on the number of requests and responses (tokens) sent and received. While we have obtained a free trial package of $18, continued use of the API will require payment based on usage. Recent Twitter API changes: Twitter has recently made changes to its API that impact the availability of certain features and data. For example, as of February 2022, Twitter has suspended access to the user profile and follower count endpoints, making it difficult to determine the size of a tweet author’s audience. This may impact the effectiveness of our blog promotion toolkit. Twitter’s recent announcement regarding automation: Twitter has announced new rules regarding automation on the platform, aimed at preventing spam and abusive behavior. While the full extent of these changes is not yet clear, they may impact the functionality of our blog promotion toolkit. We will continue to monitor developments and adjust our approach as needed. These limitations highlight the challenges involved in building and maintaining a blog promotion toolkit that relies on external APIs and platforms. Despite these limitations, I believe that our toolkit can still be effective in promoting our content and engaging with our audience. By staying informed and adapting to changes as they arise, we can continue to use these tools to achieve our goals. Final Words In conclusion, our blog promotion toolkit is designed to help content creators reach new audiences by engaging with relevant Twitter users. By leveraging the power of AI language models and Twitter’s API, we are able to generate personalized replies to tweets that mention our blog post’s relevant content in the context of the target tweet message. However, it’s important to note that there are limitations to this approach, including Twitter API rate limits and monthly search limits, as well as the cost of using OpenAI’s API for language processing. Overall, I believe that this toolkit can be a valuable asset for bloggers and content creators looking to promote their work on social media. I look forward to continuing to refine and improve this toolkit, and we welcome any feedback or suggestions from the community. Thank you for considering my toolkit, and I hope it can help you achieve your content promotion goals! Originally published at https://thesametech.com on March 3, 2023.

How to Have an Automated Blog Promoter With ChatGPT, Twitter, and AWS

Using Redis Streams with NestJS: Part 3 - Consumer groups

Apache Kafka is a distributed event streaming platform built with an emphasis on reliability, performance, and customization. Kafka can send and receive messages in a fashion. To achieve this, the ecosystem relies on few but strong basic concepts, which enable the community to build many features solving , for instance: publish-subscribe numerous use cases Processing messages as an . Enterprise Service Bus Tracking Activity, metrics, and telemetries. Processing Streams. Supporting . Event sourcing Storing logs. This article will see the concepts backing up Kafka and the different tools available to handle data streams. Architecture The behavior of Kafka is pretty simple: push into a particular , and subscribe to this to fetch and process the Let’s see how it is achieved by this technology. Producers Messages Topic Consumers Topic Messages. Infrastructure side Independently of the use, the following components will be deployed: One or more sending messages to the brokers. Producers One or more Kafka , the actual messaging server handling communication between producers and consumers. Brokers One or more fetching and processing messages in clusters named . Consumers Consumer Groups One or more instances managing the brokers. Zookeeper (Optionally) One or more instances uniformizing message schema. Registry As a scalable distributed system, Kafka is heavily relying on the concept of . As a result, on typical production deployment, there will likely be multiple instances of each component. clusters A is a cluster of the same consumer application. This concept is heavily used by Kafka to balance the load on the applicative side of things. Consumer Group Note: The dependency on Zookeeper will be removed soon, Cf. KIP-500 Applicative side A in Kafka is a pair. Those elements can be anything from an integer to a , provided the right serializer and deserializer. Message key-value Protobuf message The message is sent to a , which will store it as a . The topic should be a collection of logs semantically related, but without a particular structure imposed. A  topic can either keep every message as a new log entry or only keep the last value for each key (a.k.a. Compacted log). Topic Log To take advantage of the multiple brokers, topics are into by default. Kafka will assign any received message to one partition depending on its key, or using otherwise, which results in a random assignment from the developer's point of view. Each partition has a responsible for all I/O operations, and replicating the data. A follower will take over the leader role in case of an issue with the current one. sharded Partitions a partitioner algorithm Leader Followers The partition holds the received data in order, increasing an integer for each message. However, there is no order guarantee between two partitions. So for order-dependent data, one must ensure that they end up in the same partition by using the same key. offset Each partition is assigned to a specific consumer from the consumer group. This consumer is the only one fetching messages from this partition. In case of shutdown of one customer, the brokers will among the customers. reassign partitions Being an asynchronous system, it can be hard and impactful on the performances to have every message delivered exactly one time to the consumer. To mitigate this, Kafka provides on the number of times a message will be processed ( at most once, at least once, exactly once). different levels of guarantee i.e., Schema and Registry Messages are serialized when quitting a producer and deserialized when handled by the consumer. To ensure compatibility, both must be using the same data definition. Ensuring this can be hard considering the application evolution. As a result, when dealing with a production system, it is recommended to use a schema to explicit a contract on the data structure. To do this, Kafka provides a server, storing and binding schema to topics. Historically only was available, but the registry is now modular and can also handle and out of the box. Registry Avro JSON Protobuf Once a producer sent a schema describing the data handled by its topic to the registry, other parties ( brokers and consumers) will fetch this schema on the registry to validate and deserialize the data. i.e., Integrations Kafka provides multiple ways of connecting to the brokers, and each can be more useful than the others depending on the needs. As a result, even if a library is an abstraction layer above another, it is not necessarily better for every use case. Kafka client library There are client libraries available in , which help develop a producer and consumer easily. We will use Java for the example below, but the concept remains identical for other languages. numerous languages The producer concept is to publish messages at any moment, so the code is pretty simple. { { Properties producerProperties = Properties();
    producerProperties.put( , );
    producerProperties.put( , );
    producerProperties.put( , );
    producerProperties.put( , );
    producerProperties.put( , );
    producerProperties.put( , );
    producerProperties.put( , ); Producer<Long, AvroHelloMessage> producer = KafkaProducer<>(producerProperties); producer.send( AvroHelloMessage( , , , ));
  }
} public class Main Exception public static void main (String[] args) throws // Configure your producer new "bootstrap.servers" "localhost:29092" "acks" "all" "retries" 0 "linger.ms" 1 "key.serializer" "org.apache.kafka.common.serialization.LongSerializer" "value.serializer" "io.confluent.kafka.serializers.KafkaAvroSerializer" "schema.registry.url" "http://localhost:8081" // Initialize a producer new // Use it whenever you need new 1L "this is a message" 2.4f 1 The code is a bit more complex on the consumer part since the consumption loop needs to be created manually. On the other hand, this gives more control over its behavior. The consumer state is automatically handled by the Kafka library. As a result, restarting the worker will start at the most recent offset he encountered. { { 
    Properties consumerProperties = Properties();
    
    consumerProperties.put( , );
    consumerProperties.put( , );
    consumerProperties.put( , );
    consumerProperties.put( , );
    consumerProperties.put( , ); consumerProperties.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, ); consumerProperties;
  } { Consumer<Long, AvroHelloMessage> consumer = KafkaConsumer<>(configureConsumer()); consumer.subscribe(Pattern.compile( )); AtomicBoolean shouldStop = AtomicBoolean( );
    Thread consumerThread = Thread(() -> { Duration timeout = Duration.ofSeconds( ); (!shouldStop) { (ConsumerRecord<Long, AvroHelloMessage> record : consumer.poll(timeout)) { AvroHelloMessage value = record.value();
        } Thread.sleep( );
      }
      
      consumer.close(timeout);
    }); consumerThread.start(); shouldStop.set( );
    consumerThread.join();
  }
} public class Main Properties public static configureConsumer () new "bootstrap.servers" "localhost:29092" "group.id" "HelloConsumer" "key.deserializer" "org.apache.kafka.common.serialization.LongDeserializer" "value.deserializer" "io.confluent.kafka.serializers.KafkaAvroDeserializer" "schema.registry.url" "http://localhost:8081" // Configure Avro deserializer to convert the received data to a SpecificRecord (i.e. AvroHelloMessage) // instead of a GenericRecord (i.e. schema + array of deserialized data). true return Exception public static void main (String[] args) throws // Initialize a consumer final new // Chose the topics you will be polling from. // You can subscribe to all topics matching a Regex. "hello_topic_avro" // Poll will return all messages from the current consumer offset final new false new final 5 while for // Use your record // Be kind to the broker while polling 5 // Start consuming && do other things // [...] // End consumption from customer true Kafka Streams Kafka Streams is built on top of the consumer library. It continuously reads from a topic and processes the messages with code declared with a functional DSL. During the processing, transitional data can be kept in structures called and , which are stored into topics. The former is equivalent to a standard topic, and the latter to a compacted topic. Using these data stores will enable automatic tracking of the worker state by Kafka, helping to get back on track in case of restart. KStream KTable The following code sample is extracted from the . The code connects to a topic named containing strings values, without necessarily providing keys. The few lines configuring the will: tutorial provided by Apache streams-plaintext-input StreamsBuilder Transform each message to lowercase. Split the result using whitespaces as a delimiter. Group previous tokens by value. Count the number of tokens for each group and save the changes to a KTable named . counts-store Stream the changes in this Ktable to send the values in a KStream named output. streams-wordcount- { {
    Properties props = Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, );
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, );
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); StreamsBuilder builder = StreamsBuilder();

    builder.<String, String>stream( )
            .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split( )))
            .groupBy((key, value) -> value)
            .count(Materialized.<String, Long, KeyValueStore<Bytes, []>>as( ))
            .toStream()
            .to( , Produced.with(Serdes.String(), Serdes.Long())); Topology topology = builder.build(); KafkaStreams streams = KafkaStreams(topology, props); CountDownLatch latch = CountDownLatch( ); Runtime.getRuntime().addShutdownHook( Thread( ) { {
        streams.close();
        latch.countDown();
      }
    }); streams.start();
    latch.await();
  }
} public class Main Exception public static void main (String[] args) throws new "streams-wordcount" "localhost:29092" final new "streams-plaintext-input" "\\W+" byte "counts-store" "streams-wordcount-output" final final new final new 1 // attach shutdown handler to catch control-c new "streams-shutdown-hook" @Override public void run () // The consumer loop is handled by the library Kafka Connect Kafka Connect provides a way of transforming and synchronizing data between almost any technology with the use of . Confluent is hosting a , on which users can share connectors for various technologies. This means that integrating a Kafka Connect pipeline is most of the time only a matter of configuration, without code required. A single connector can even handle both connection sides: Connectors Hub Populate a topic with data from any system: a . i.e. Source Send data from a topic to any system: a . i.e. Sink The source will read data from CSV files in the following schema then publish them into a topic. Concurrently, the sink will poll from the topic and insert the messages into a MongoDB database. Each connector can run in the same or a distinct worker, and workers can be grouped into a cluster for scalability. The connector instance is created through a configuration specific to the library. The file below is a configuration of the MongoDB connector. It asks to fetch all messages from the topic mongo-source to insert them into the collection sink of the database named kafka-connect. The credentials are provided from an external file, which is a feature of Kafka Connect to protect secrets. { : , : { : , : , : , : , : , : , : , : , : , : , : , : }
} "name" "mongo-sink" "config" "topics" "mongo-source" "tasks.max" "1" "connector.class" "com.mongodb.kafka.connect.MongoSinkConnector" "connection.uri" "mongodb://${file:/auth.properties:username}:${file:/auth.properties:password}@mongo:27017" "database" "kafka_connect" "collection" "sink" "max.num.retries" "1" "retries.defer.timeout" "5000" "document.id.strategy" "com.mongodb.kafka.connect.sink.processor.id.strategy.BsonOidStrategy" "post.processor.chain" "com.mongodb.kafka.connect.sink.processor.DocumentIdAdder" "delete.on.null.values" "false" "writemodel.strategy" "com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneDefaultStrategy" Once the configuration complete, registering the connector is as easy as an HTTP call on the running Kafka Connect instance. Afterward, the service will automatically watch the data without further work required. http://localhost:8083/connectors -d @sink-conf.json $ curl -i -X POST -H -H \ "Accept:application/json" "Content-Type:application/json" KSQL Database Ksql is somehow equivalent to Kafka Streams, except that every transformation is declared in an SQL-like language. The server is connected to the brokers and can create or from topics. Those two concepts behave in the same way as a KStream or KTable from Kafka Streams ( respectively, a topic and a compacted topic). Streams Tables i.e., There are three types of queries in the language definition: ( ): Creates a new stream or table that will be automatically updated. Persistent Query e.g. CREATE TABLE <name> WITH (...) ( ): Behaves similarly to a standard DBMS. Fetches data as an instant snapshot and closes the connection. Pull Query e.g. SELECT * FROM <table|stream> WHERE ID = 1 ( ): Requests a persistent connection to the server, asynchronously pushing updated values. Push Query e.g. SELECT * FROM <table|stream> EMIT CHANGES The database can be used to browse the brokers' content. Topics can be discovered through the command , and their content displayed using . list topics print <name> ksql> list topics;
 Kafka Topic      | Partitions | Partition Replicas hello_topic_json | 1          | 1 ksql> print 'hello_topic_json' from beginning;
Key format: KAFKA_BIGINT or KAFKA_DOUBLE or KAFKA_STRING
Value format: JSON or KAFKA_STRING
rowtime: 2021/05/25 08:44:20.922 Z, key: 1, value: {"user_id":1,"message":"this is a message","value":2.4,"version":1}
rowtime: 2021/05/25 08:44:20.967 Z, key: 1, value: {"user_id":1,"message":"this is another message","value":2.4,"version":2}
rowtime: 2021/05/25 08:44:20.970 Z, key: 2, value: {"user_id":2,"message":"this is another message","value":2.6,"version":1} ---------------------------------------------------- ---------------------------------------------------- The syntax to create and query a stream or a table is very close to SQL. ksql> messages (user_id PRIMARY , message ) 
> (KAFKA_TOPIC = , VALUE_FORMAT= ); ksql> list tables;
 Table Name | Kafka Topic      | Key Format | Value Format | Windowed MESSAGES   | hello_topic_json | KAFKA      | JSON         | false ksql> messages;
Name                 : MESSAGES
 Field   | Type USER_ID | BIGINT           (primary key)
 MESSAGE | VARCHAR(STRING) For runtime statistics and query details run: <Stream, >; ksql> * messages EMIT CHANGES;
+ |USER_ID |MESSAGE                 |
+ |1       |this is another message |
|2       |this is another message | -- Let's create a table from the previous topic CREATE TABLE BIGINT KEY VARCHAR WITH 'hello_topic_json' 'JSON' -- We can see the list and details of each table ---------------------------------------------------------------------- ---------------------------------------------------------------------- describe ------------------------------------------ ------------------------------------------ DESCRIBE EXTENDED Table -- Appart from some additions to the language, the queries are almost declared in standard SQL. select from --------+------------------------+ --------+------------------------+ Kafka recommends using a headless ksqlDB server for production, with a file declaring all streams and tables to create. This avoids any modification to the definitions. Note: ksqlDB servers can be grouped in a cluster like any other consumer. Conclusion This article gives a broad view of the Kafka ecosystem and possibilities, which are numerous. This article only scratches the surface of each subject. But worry not, as they are all well documented by Apache, Confluent, and fellow developers. Here are a few supplementary resources to dig further into Kafka: (Youtube) Kafka Tutorials - Confluent Kafka Tutorials in Practice Top 5 Things Every Apache Kafka Developer Should Know — Bill Bejeck Kafkacat user Guide Troubleshooting KSQL Part 2: What’s Happening Under the Covers? — Robin Moffatt Apache Kafka Internals — sudan The complete experimental code is available on my GitHub repository . Also published on GitHub .

The Apprentice's Guide to Apache Kafka

5 Things Every Apache Kafka Dev Needs To Know: A Performance and Architectural Deep Dive

In this article, we’ll be discussing everything you need to know about the basics of AWS Lambda error handling and some popular methods using StepFunctions and X-Ray. Regardless if you’re an AWS Lambda expert or if you’re a new Lambda user, . You’ve probably already encountered Lambda errors that may seem pretty challenging since the mechanism that runs Lambda retries will often make it incredibly difficult to follow up on changes that occur within your serverless application. there’s always something new to learn Serverless is not all about straightforward execution of code on Lambda function, but it’s a . Distributed nodes within this architecture that are activated thanks to asynchronous events are what makes this system. different type of architecture of your entire system Every node has to be designed like a singular part with its own API. To learn how to define all these nodes accurately, you have to know . In addition, it’s necessary to deal appropriately with Lambda retry behavior as well. how to handle Lambda errors Therefore, let’s jump right into , as well as what’s the whole buzz around it all. how do the AWS Lambda retries and errors work Lambda Retry Behavior Lambda functions can fail ( ), and when they do, it’s because of one of these situations: they will — if you’re out of memory, Lambda often terminates with the message The is always the same as You can learn more about resource allocation and AWS Lambda memory here. Lack of memory ‘Process exited before completing the request.’ ‘Memory Size’ ‘Max Memory Used.’ — can happen because of a programming bug, failure of an external API, or if you’ve received an invalid input. Raised unhandled exception — message appears when Lambda closes violently because it ran longer than the pre-configured timeout duration. The maximal value is five minutes, while the default value is set at six seconds. Timeout ‘Task timed out after X seconds’ When failure occurs, and it will occur at some point, you’ll most likely notice : Lambda retries based on these behaviors – if the current events are solely DynamoDB streams and AWS Kinesis Data Streams. When it happens, AWS triggers these failing Lambda functions again until they’re processed successfully or until the data expires, and AWS will block the event’s source until it happens. Stream-based events – In event sources like synchronous invocation or API Gateway using the SDK, the invoked app is responsible for creating response-based retries that it gets from . This case scenario is the least interesting one since it resembles monolithic error handling. Synchronous events Lambda – Lambda invocation happens asynchronously for most event sources. This means that no app will respond to failure. Therefore, the . It’ll trigger the Lambda again with precisely the same event, which happens mostly twice within the upcoming ~3 minutes (in some sporadic cases, it takes up even to six hours, while a different number of continuous retries can happen). Asynchronous events AWS framework will take care of it on its own In case all retries fail, it’s mandatory to record this event instead of throwing it away. That’s why the crucial Dead Letter Queue (DLQ) feature enables the configuration of DLQ via AWS SQS that’ll receive this type of event.Consequences of AWS Lambda Retry Behavior Every Lambda can be , while the didn’t know about it happening. To successfully perform the same operation several times, Lambda has to be ‘idempotent.’ executed multiple times with the same input “caller” means that no added effects will take place when it’s run by the same input more than one time. Idempotency It’s worth mentioning that . A standard model is the API network: when a request doesn’t receive a response, the same request will be sent repeatedly. serverless functions aren’t the only example when it comes to idempotency terms For example, in Serverless architectures, it can come to a similar case when Lambda gets a timeout before receiving a response. Even though it’s a highly unexpected thing to happen, in some cases, a wrong retry handling can be the cause of severe problems (a database (DB) structural violation). What is AWS Lambda Idempotency? The definition of idempotency states that it’s the property of specific operations in computer science and mathematics. It’s applicable several times without disrupting the result beyond the first application. However, it’s still somewhat confusing. What happens if you wish to execute the same operation several times and it’s not actually a retry? Let’s say that Lambda received a user operation log as input, and it’s solely responsible for recording that operation log into a database. In this example, we’ll need to make a difference between Lambda’s trigger input and a retry case since they’re the same because the user has initiated the same operation again. Only when there’s a Lambda retry, you’ll obtain the same ID. To be able to extract it, utilize within Node.js or the appropriate field in any other language. What this method does is it Referring to the Lambda’s request ID as a part of the input itself is the right solution. context.awsRequestId provides the general approach that’ll search for retry executions. It’s not always a convenient solution to utilize the request ID for being a genuine idempotent. You might have noticed within the previous example that this ID should’ve also been saved in the DB. That way, the following invocations would realize whether there’s a need to add a new record. There’s one more solution, and it’s . However, it might add quite a significant overhead. using some in-memory data store Using AWS Step Functions to Build a Control System for AWS Lambda Error Handling AWS Lambda error handling can be done in different ways, like utilizing wrappers. On the other hand, have proved to be for building a serverless application that’ll deal with retries and errors appropriately, making Step Functions an effective solution. You can learn more about AWS Step Functions in our Ultimate Guide to Step Functions. AWS Step Functions incredibly beneficial Take the Next Step Let’s say that the . By combining them all with the same Lambda, the . application has to perform multiple operations in response to an event code will have to check for every operation separately If you’re trying to keep your Lambda idempotent, should it be redone? Remember that this can cause severe headaches. It would help if you learned the difference between monolithic applications and the Step Function example we’ve mentioned. In a monolithic application, the because it’s capable of waiting between them, and that’s something that . application itself can become responsible for forcing retries isn’t possible in Serverless However, with Step Functions, you’re able to run every operation on separate Lambda. In addition, you’re able to define suitable transitions between them for each specific case. You can also control the retries’ behavior – their delay duration as well as their number. That way, you’ll quickly adjust it to be the most suitable one for your particular case. You’ll even disable it when you believe that’s the right step to take. Even if needed by a single Lambda, creating a step machine is possibly the most straightforward solution for disabling unwanted retry behavior. How to Implement Step Functions to Lambda? As you probably know, all available ; the only triggers that are available are API Gateway, including a manual execution utilizing SDK. Step Function triggers are quite limited To successfully deploy this Lambda, you have to , along with the incredible plugin so it could easily . Additionally, you have to ensure that you use and so you’d be capable of defining the state machine like it’s shown in the below example: utilize the Serverless framework ‘serverless-resources-env’ pass the state machine ARN ‘serverless-pseudo-parameters’ ‘serverless-step-functions’ You can see that the artificial choice of implementing an SNS event is made , and it’s accessible as input by the initial step Lambda. Everything will become idempotent since we named the state machine’s execution as the invoker In case a retry happens to this invoker Lambda, purposely to trigger the state machine deliberately ‘Lambda request ID.’ AWS will give it the same request ID. After that, AWS won’t be executing the state machine again because it has the same name. In theory, the state machine’s execution name is a part of its input as well. Even though this solution is advantageous in numerous case scenarios, you should know that it’ll also add a significant complexity overhead, further affecting the system’s overall observability and debugging. Things to Note Regarding Step Functions Error Handling Mechanism It’s essential to comprehend that the Step Function’s error handling mechanism is quite different than the AWS Lambda error handling mechanism. For each Task state, a timeout duration placeholder can be set, and in case the Task isn’t completed in time, error will be generated. This particular timeout is unlimited in a way. anStates.Timeout Also, in a typical case of a Task that executes a Lambda, the case won’t be the same. Lambda’s actual timeout duration can be determined solely by its pre-configured value, and it can’t get any longer by utilizing this method. Therefore, it’s essential to ensure that you’ve configured the Task timeout so it’d be equal to the timeout of Lambda. The Task’s retry behavior is disabled by default, and it can be configured in a certain way. AWS Lambda Error Processor Sample Application The Error Processor sample application shows the utilization of AWS Lambda to handle events that are coming from the AWS CloudWatch Logs subscription. Now, function if a . The subscription within this application will monitor a function’s log group for all entries with the word . In response, it’ll The processor function will then retrieve the full log-stream and trace data for the request that have caused this error, and it’ll store them so it could use them later. CloudWatch Logs will allow you to invoke a Lambda log entry matches a particular pattern ERROR invoke a processor Lambda function. Function code can be found in these files: Processor – processor/index.jsRandom error – random-error/index.js You can quickly deploy the sample within a few minutes via AWS CloudFormation and AWS CLI. Event Structure and Architecture This sample application utilizes these AWS services: – which will store application output and deployment artifacts. Amazon S3 – collect logs, but when a log entry matches a filter pattern, it’ll also invoke a function. Amazon CloudWatch Logs – Generates a service map, indexes traces needed for search, and collects trace data. AWS X-Ray – Sends all the trace data to the X-Ray, sends logs to CloudWatch Logs, and runs a function code. AWS Lambda A Lambda function will when found within the application. If CloudWatch Logs detect the word within the function’s logs, it’ll provide the processor function with an event for processing. generate errors randomly ERROR CloudWatch Logs message event The data has details about the log event when it’s decoded. The function will use all these details to successfully and so it would obtain the ID of the particular request that has caused this error. identify the log stream parses the log message Decoded CloudWatch Logs event data The processor function will utilize information obtained from the CloudWatch Logs event to download the X-Ray trace and the full log stream for a request Both will be Moreover, to successfully allow the trace time and log stream to finalize, the function will wait for a short period before it starts accessing the data. that has caused an error. stored within the AWS S3 bucket. AWS X-Ray Instrumentation The application utilizes AWS X-Ray to t and all the X-Ray utilizes the received trace data from functions to create a that is of significant help for This particular service map showcases the random error function that generates errors for some specific requests. Additionally, it showcases the processor function that calls CloudWatch Logs, Amazon S3, and X-Ray. race function invocations calls that functions make to AWS. service map error identification. These two configured Node.js functions serve within the template and are instrumented with the AWS X-Ray SDK (Node.js) in code. Along with active tracing, Lambda tags will add a to all incoming requests, and they’ll send a trace with to AWS X-Ray. Moreover, the random error function utilizes X-Ray SDK to record the request ID and the user information within annotations. These annotations are attached to the trace, so active tracing tracing header timing details you could use them to locate the specific request’s trace. The processor function will obtain the request ID from the CloudWatch Logs event, and it’ll utilize the AWS SDK for JavaScript to search X-Ray for that particular request. It also utilizes , which are instrumented by the X-Ray SDK to download the log stream and the trace. After that, it’ll store them in the output bucket. The X-Ray SDK will record all these calls, and they’ll appear within the trace as subsegments. AWS SDK clients AWS CloudFormation Template and Additional Resources The application is implemented within the two Node.js modules, and it’s deployed with shell scripts and AWS CloudFormation template. This template will create the , , and all the following : processor function the random error function supporting resources – is an IAM role that allows function with permission to access other AWS services. – is an additional function whose objective is to invoke the random error function to create a specific log group. – is another AWS CloudFormation custom resource that’ll invoke the primer function during deployment, so it would make sure that this particular log group exists. – is a subscription for the log stream triggering the processor function when word is successfully logged. – is a specific permission statement on the processor function which allows invocation via CloudWatch Logs. – is an output storage location (processor function). Execution role Primer function Custom resource CloudWatch Logs subscription ERROR Resource-based policy Amazon S3 bucket If you’re trying to with CloudFormation successfully, the template will come up with an that’ll . All Lambda functions will always come with a specific CloudWatch log group that’ll . Additionally, this log group won’t be created until the function has been invoked for the first time. work around the limitations of integrating Lambda additional function run during deployments store the output from all function executions Creating a subscription that depends on the log group’s sole existence, the application needs to utilize a 3rd Lambda function to invoke the random error function. This template also includes the primer function inline code. Each AWS CloudFormation custom resource is capable of invoking it during its deployment. properties will ensure that the and the are DependsOn resource-based policy log stream created before the subscription. How Dashbird can help handle your AWS Lambda Errors faster and easier fundamentally change how we develop, deploy, and monitor applications. As you now know, services such as AWS Lambda also come with their own limits and idiosyncrasies: limited memory and execution time, retry-behavior, and many others may create side-effects that can easily become . Serverless architectures monitoring nightmares Composing multiple services for compute, data storage, queues, etc. magnifies the problem. The number of potential issues is multiplied by the interactions and dependencies throughout the cloud stack. Running such architectures at scale is even more challenging. At each level of traffic, we cannot expect the stack will behave homogeneously. Perhaps AWS Lambda functions will scale faster than a database, for example. Dashbird is designed to while achieving a high degree of visibility and quality in any serverless architecture. provide developers with ways to easily navigate such complex problems to enhance monitoring and operating specifically for AWS services at scale. By , Dashbird all your code exceptions, timeouts, configuration errors, and other anomalies , and sends you a immediately if there’s an error or something is about to break. Dashbird was created by serverless developers, for serverless developers continuously collecting and filtering your log data automatically detects in real-time notification Wrapping up AWS Lambda error handling in serverless architecture may seem pretty confusing, but as much as it’s hard to comprehend how it can affect your entire system, it’s vital to thoroughly understand it. It’s important that you know how to , and the . Every retry counter field within the context parameter is undoubtedly a feature that’s been missing. manage AWS Lambda retries behavior successfully same goes for Step Functions Besides the techniques mentioned in this article, there are various other methods that will help with AWS Lambda error handling, and utilizing wrappers is only one example. The architecture with Step Functions that we’ve discussed today is quite useful in many cases, and AWS Lambda error handling is one of them. Even though it helps control Lambda retries appropriately, it also encourages the , which is an excellent practice within the world of Serverless. separation of elements Also published at: https://dashbird.io/blog/aws-lambda-error-handling-step-functions/

TIME OUT GROUP

timeout.com

201-500 employees

Since 1968

COMPANY RANKING

TIME OUT GROUP

EVERGREEN INDEX #10731

Representing this company? Click here to claim and customize this page!

Time Out Group WIKI

Apply to get your Tech Company News Page live on HackerNoon.com today!

TIME OUT GROUP

timeout.com

201-500 employees

Since 1968

COMPANY RANKING

TIME OUT GROUP

EVERGREEN INDEX #10731

RELATED COMPANIESRANK

STORIES ABOUT TIME OUT GROUP

Representing this company? Click here to claim and customize this page!

Time Out Group WIKI

Apply to get your Tech Company News Page live on HackerNoon.com today!