DEFAULT_KEY_SERDE_CLASS_CONFIG in Kafka

The DEFAULT_KEY_SERDE_CLASS_CONFIG is a configuration property in Apache Kafka Streams. It specifies the default serializer/deserializer (Serde) to be used for the keys of messages in your Kafka Streams application.

What Is a Serde?

A Serde is a combination of a serializer and a deserializer:

  • Serializer: Converts objects into a byte array to send over Kafka.
  • Deserializer: Converts the byte array back into an object.

Kafka Streams requires Serdes to know how to handle data when reading from and writing to topics.

Usage of DEFAULT_KEY_SERDE_CLASS_CONFIG

This property allows you to specify the default Serde class for the key of messages processed by Kafka Streams. For example, if your keys are strings, you would set the default key Serde to Serdes.String().

Example Configuration in Java:

Properties props = new Properties();
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

Example Configuration in application.yml:

spring:
  kafka:
    streams:
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde

Common Serdes in Kafka Streams

Kafka Streams provides built-in Serdes for common data types:

  • StringSerde: For string keys/values.
    org.apache.kafka.common.serialization.Serdes.StringSerde
  • IntegerSerde: For integer keys/values.
    org.apache.kafka.common.serialization.Serdes.IntegerSerde
  • LongSerde: For long keys/values.
    org.apache.kafka.common.serialization.Serdes.LongSerde
  • DoubleSerde: For double keys/values.
    org.apache.kafka.common.serialization.Serdes.DoubleSerde
  • ByteArraySerde: For raw binary data.
    org.apache.kafka.common.serialization.Serdes.ByteArraySerde

Why Use DEFAULT_KEY_SERDE_CLASS_CONFIG?

  1. Convenience: Sets a global default, so you don't need to specify the Serde for every stream/table.
  2. Consistency: Ensures that all keys are handled the same way unless overridden.

Overriding the Default Serde

If you need to process data differently for a specific stream, you can override the default Serde when defining the stream:

KStream<Integer, String> stream = builder.stream(
    "source-topic",
    Consumed.with(Serdes.Integer(), Serdes.String()) // Custom key/value Serdes
);

Let me know if you'd like to explore custom Serdes!

댓글

댓글 쓰기

이 블로그의 인기 게시물

What are 5 creative things I could do with my kids' art, 아이와 함께하는 창의적 놀이

Absolutely, preserving your kids' art while also finding creative ways to utilize and display it can be a wonderful solution. Here are five creative ideas to consider: Art Gallery Wall: Designate a section of a wall in your home as an art gallery. Frame selected pieces of your kids' art and arrange them in an organized or eclectic manner. You can change the display periodically to showcase different artworks, creating a rotating gallery. Customized Photo Books: Compile your children's art into a digital photo book or album. Many online services allow you to create customized photo books. Arrange the artwork, add captions, and create a keepsake that you can flip through whenever you want to reminisce. Transform into Functional Items: Turn your kids' artwork into functional items such as placemats, coasters, tote bags, or even custom-made wrapping paper. There are companies that can print your children's art onto these items, making them practical and sentime...
자세한 내용 보기

how to make inactive kafka topic, 카프카 토픽 비활성화

To make a Kafka topic inactive, meaning it no longer receives new messages and its data is eventually deleted, there are several strategies you can follow. Kafka doesn’t have a specific “inactive” state for topics, but you can achieve this effect by adjusting the configuration settings and permissions. Here are some methods to make a Kafka topic "inactive": 1. Restrict Producer Access (Prevent Writing New Messages) One way to make a Kafka topic inactive is to prevent producers from writing new messages to it. You can do this by modifying access control lists (ACLs) or configurations, depending on your Kafka setup. Steps: Modify ACLs to restrict producer access to the topic. If you have security enabled, you can revoke the producer’s permission to write to the topic. bin/kafka - acls . sh - - bootstrap - server localhost:9092 - - remove - - allow - principal User: < producer - user > - - operation Write - - topic < topic - name > This command will ...
자세한 내용 보기

Screenshot of a web page using Puppeteer with a specific browser width, 웹페이지 스크린샷

To get a screenshot of a web page using Puppeteer with a specific browser width, you can follow these steps: Install Puppeteer by running the command npm install puppeteer in your project directory. Require Puppeteer in your Node.js script: const puppeteer = require( 'puppeteer' ); Launch a new browser instance with the desired viewport size: const browser = await puppeteer.launch({ defaultViewport: { width : 1280 , height : 720 , }, }); Create a new page in the browser: const page = await browser. newPage () ; Navigate to the desired URL: await page. goto ( 'https://example.com' ); Take a screenshot of the page: await page .screenshot ({ path : 'screenshot.png' }); Close the browser instance: await browser.close() ; Here's the complete code: const puppeteer = require ( 'puppeteer' ); ( async ( ) => { const browser = await puppeteer.launch({ defaultViewport : { width : 1280 , ...
자세한 내용 보기

리눅스의 부팅과정 (프로세스, 서비스 관리)

보통 사용되지 않는 서비스나 사용 빈도가 적은 프로세스의 경우 stop 시켜두는 것이 시스템 사용 상의 이점이 있다. 자신에게 필요 없는 프로세스를 정지 한다거나 nice 값을 조정하여 실행 시에 프로세스의 우선순위를 변화시킬 수 있다. 이렇게 시스템 상황에 따라 적절히 서비스들을 관리하려면 현재 자신의 시스템의 여러가지 상황을 알아야 하는데, 이런 경우 프로세스 및 실시간 시스템 상황을 보는 명령어인 top 을 사용할 수 있다.  top [-][d delay][q][c][S][s][i][n][b] -d delay : delay 의 시간이 경과하면 현재 스크린을 갱신한다. -q : 이 옵션을 사용하면 스크린을 계속 갱신한다. -c : command list 전체를 보여준다. 즉 욥선을 사용한 것까지 모두 보여준다. -i : idle 상태와 zombie 프로세스는 무시한다.   * 이 외의 옵션은 man 파일을 참고 top 실행 화면 1. First Line :  2:59am up 6days, 10:36,  4 users,  load average    : 현재 시간 2:59 am , 부팅된지 6days and 10시간 36분, 접속해 있는 사용자수, 평균부하 2. Second Line : 61 processes: 56 sleeping, 3 running, 1 xombie, 1 stopped 3. Third Line : CPU states: 36.4% user, 63.5% system, 0.0% nice, 0.0% idle     유저모드에서의 CPU 시간이 36.4%     시스템모드(프로세스를 위해서 커널이 사용한 CPU의 시간)에서의 CPU시간이 63.5%     0.0% nice는 nice로 nice value를 음수로 주어 우선순위를 높이는 경우에 해당하는 모드     0.0% idle(...
자세한 내용 보기

To change the ESE index in Fluent Bit

To change the ESE index in Fluent Bit, you can follow these steps: Open the Fluent Bit configuration file (typically located at /etc/td-agent-bit/td-agent-bit.conf or /etc/fluent-bit/fluent-bit.conf , depending on your installation). Locate the output section for the ESE output plugin. It should look something like this: [OUTPUT] Name ese Match ese.* Index fluent-bit Database /var/ lib / td - agent - bit / ese . db Interval_Sec 10 Change the value of the Index parameter to the desired name of your ESE index. For example, if you want to create an index named "myindex", you would change the configuration to: [OUTPUT] Name ese Match ese.* Index myindex Database /var/ lib / td - agent - bit / ese . db Interval_Sec 10 Save the configuration file and restart the Fluent Bit service to apply the changes. On Linux, you can restart Fluent Bit using the following command: sudo systemctl ...
자세한 내용 보기

Consume a topic from kafka by Python

Sure, I can help you develop Python code to consume topics from Apache Kafka using the confluent-kafka library. Make sure you have Kafka installed and running, and you have the confluent-kafka library installed using pip install confluent-kafka . Here's an example of how you can consume messages from a Kafka topic: from confluent_kafka import Consumer, KafkaError # Kafka configuration config = { 'bootstrap.servers' : 'localhost:9092' , # Replace with your Kafka broker addresses 'group.id' : 'my-group' , # Consumer group ID 'auto.offset.reset' : 'earliest' # Start consuming from the beginning of the topic } # Create Kafka consumer instance consumer = Consumer(config) # Subscribe to a topic topic = 'my-topic' # Replace with the topic you want to consume from consumer.subscribe([topic]) # Consume messages try : while True : msg = consumer.poll( 1.0 ) # Poll for new messages with a...
자세한 내용 보기

ssh-copy-id in ansible

ssh-copy-id is a command-line tool used to copy your public SSH key to a remote server's authorized_keys file, allowing you to log in to the server without entering a password. Ansible is an open-source automation tool used to manage and configure multiple remote servers simultaneously. In Ansible, you can use the authorized_key module to manage the authorized keys of remote servers. This module allows you to add, remove, or replace SSH keys in a remote user's authorized_keys file. To add your public SSH key to a remote server using Ansible, you can use the authorized_key module. Here's an example playbook that adds your public SSH key to a remote server's authorized_keys file: - name: Add SSH key to authorized_keys hosts: your_server become: true tasks: - name: Install SSH key for current user authorized_key: user: your_username key: " {{ lookup('file', '/path/to/your/public/key/id_rsa.pub') }} " ...
자세한 내용 보기

패션 하의, Pants, Skirts, Shorts, Skorts

하의의 종류는 다양한 디자인, 용도, 계절 등에 따라 나뉩니다. 대표적인 하의 종류를 설명하자면 다음과 같습니다. 1. 바지 (Pants) 청바지 (Jeans) : 데님 소재로 만들어진 바지로, 캐주얼한 스타일의 기본 아이템입니다. 청바지는 다양한 워싱, 핏으로 제공되며 일상적인 스타일에 많이 사용됩니다. 슬랙스 (Slacks) : 정장이나 격식 있는 자리에 입는 바지로, 주로 직물로 만들어집니다. 깔끔한 실루엣을 제공하며 포멀한 상황에 적합합니다. 조거 팬츠 (Jogger Pants) : 발목 부분이 밴드로 조여지는 캐주얼한 스타일의 바지로, 편안한 착용감을 제공합니다. 운동복으로도 자주 사용됩니다. 레깅스 (Leggings) : 몸에 딱 맞는 신축성 있는 재질로 만들어진 바지로, 주로 운동복이나 일상적인 편안한 룩에 사용됩니다. 2. 스커트 (Skirts) 미니 스커트 (Mini Skirt) : 무릎 위의 길이로 짧은 스타일의 스커트로, 캐주얼하거나 세련된 느낌을 줍니다. 미디 스커트 (Midi Skirt) : 무릎 아래에서 종아리 중간까지 오는 길이의 스커트로, 적당히 단정하면서도 여성스러운 느낌을 줍니다. 맥시 스커트 (Maxi Skirt) : 발목까지 내려오는 긴 스커트로, 여유롭고 편안한 스타일입니다. 플리츠 스커트 (Pleated Skirt) : 여러 개의 주름이 있는 스커트로, 클래식한 스타일과 현대적인 스타일 모두에서 사용됩니다. 3. 반바지 (Shorts) 데님 반바지 (Denim Shorts) : 청바지 재질로 만들어진 짧은 바지로, 주로 여름철에 입습니다. 치노 반바지 (Chino Shorts) : 면 소재로 만들어진 반바지로, 캐주얼하면서도 깔끔한 느낌을 줍니다. 버뮤다 반바지 (Bermuda Shorts) : 무릎 길이 정도의 반바지로, 편안하면서도 단정한 느낌을 줄 수 있습니다. 4. 치마바지 (Skorts) 치마처럼 보이지만 속에 반바지가 내장된 형태의 하의로, 활동성이 높고 캐주얼하게 입을...
자세한 내용 보기

ftplib.error_perm: 550 Permission denied

The 550 Permission denied error in ftplib occurs when the FTP server denies access to the specified file or directory. This error can be caused by several issues such as: Incorrect file path or directory path on the server. Lack of write permissions for the specified directory. The FTP user does not have the required permissions to upload to the given directory. The remote directory may not exist. Steps to Troubleshoot and Resolve the Issue Here are some steps you can try to fix the 550 Permission denied issue: 1. Check the File Path and Directory on the FTP Server Make sure the remote path you are trying to upload to exists on the FTP server. If you're uploading to a specific directory, you may need to create it first. You can use the following code to list the files and directories on the server to ensure your paths are correct: from ftplib import FTP def list_files (ftp_host, ftp_user, ftp_password) : try : ftp = FTP(ftp_host) ftp.login(...
자세한 내용 보기