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!

댓글

댓글 쓰기

이 블로그의 인기 게시물

To build a gRPC service with Gradle, Java, and Avro

To build a gRPC service with Gradle , Java , and Avro , follow these steps. This example will walk you through creating a basic user data service that allows clients to send and retrieve user information. Project Structure grpc-avro-user-service/ │ ├── src/ │ ├── main/ │ │ ├── avro/ │ │ │ └── user .avsc │ │ ├── java/ │ │ │ ├── proto/ │ │ │ │ └── UserService .proto │ │ │ └── com/example/userservice/ │ │ │ ├── UserServiceImpl .java │ │ │ └── App .java ├── build .gradle └── settings.gradle Step-by-Step Guide 1. Define the Avro Schema ( user.avsc ) Create an Avro schema for user data, which will be serialized and used by the service. src/main/avro/user.avsc : { "namespace" : "com.example.userservice" , "type" : "record" , "name" : "User" , "fields" : [ { "name" : "id" , "type" : "string" }, { ...
자세한 내용 보기

The pierce selector in Puppeteer

The pierce selector in Puppeteer is an experimental feature that allows you to query shadow DOM elements more easily. Traditional CSS selectors don't pierce through shadow roots because they encapsulate their content to prevent styles and scripts from leaking in or out. The pierce selector solves this by bypassing the shadow DOM boundary and directly selecting elements within shadow trees. How to Use the Pierce Selector in Puppeteer Here’s an example of how to use it effectively: const puppeteer = require ( 'puppeteer' ); ( async ( ) => { const browser = await puppeteer.launch({ headless : false }); const page = await browser.newPage(); // Navigate to a page with shadow DOM await page.goto( 'https://example.com' ); // Select an element inside a shadow DOM using the pierce selector const element = await page.$( 'pierce:#shadow-element-id' ); if (element) { console .log( 'Shadow element found!' ); awai...
자세한 내용 보기

how to describe kafka topics, 카프카 토픽 정보 확인하기

To describe Kafka topics, you can use the kafka-topics.sh script, which is included in the Kafka distribution, or Kafka Admin APIs to programmatically fetch topic information. Here are the ways to describe Kafka topics: 1. Using kafka-topics.sh Script The kafka-topics.sh command-line tool allows you to describe topics on the Kafka cluster. You can use the --describe flag to get details like the number of partitions, replicas, leader, and ISR (in-sync replicas). Command Syntax: bin/kafka - topics . sh - - describe - - topic < topic - name > - - bootstrap - server < broker - address > Example: bin/kafka - topics . sh - - describe - - topic my - topic - - bootstrap - server localhost:9092 Output Example: Topic: my-topic PartitionCount: 3 ReplicationFactor: 2 Configs: segment.bytes= 1073741824 Topic: my-topic Partition: 0 Leader: 1 Replicas: 1 , 2 Isr: 1 , 2 Topic: my-topic Partition: 1 Leader: 2 Replicas: 2 , 1 Is...
자세한 내용 보기

Using the MinIO API via curl

Using the MinIO API via curl is straightforward, as MinIO is compatible with Amazon S3 API, so most commands follow a similar syntax. Here’s a guide on how to use curl with the MinIO API for some common operations like uploading, downloading, and managing objects. Prerequisites Access Key and Secret Key : Obtain your MinIO Access Key and Secret Key. MinIO Endpoint : Know your MinIO server endpoint, e.g., http://localhost:9000 . Bucket : You may need an existing bucket name, or create a new one using the commands below. Authentication Header For requests to work with MinIO, you need to include authentication in the headers. MinIO uses AWS Signature Version 4 for signing requests. Common Examples 1. List Buckets To list all buckets in your MinIO account, use: curl -X GET \ - -url "http://localhost:9000/" \ - H "Authorization: AWS <AccessKey>:<Signature>" 2. Create a Bucket To create a new bucket, use: curl -X PUT \ - -url "htt...
자세한 내용 보기

오늘의 문장2

  I subscribed to a weekly newsletter about upcoming events, which often include concerts and exhibitions that I don't want to miss. 나는 놓치고 싶지 않은 콘서트와 전시회를 자주 포함하는 다가오는 행사에 관한 주간 뉴스레터를 구독했다. We'll update the flight information as soon as the airline confirms the schedule, so that passengers can adjust their plans accordingly. 항공사가 일정을 확인하는 대로 우리는 항공편 정보를 업데이트할 것이며, 승객들은 그에 맞춰 계획을 조정할 수 있다. Police uphold the rule of law, even when facing public criticism, because they believe justice must be maintained at all times. 경찰은 항상 정의가 유지되어야 한다고 믿기 때문에, 대중의 비판에 직면하더라도 법치를 지킨다. He stood upright on the deck, which was swaying with the waves, refusing to give in to fear. 그는 파도로 흔들리는 갑판 위에 똑바로 서서 두려움에 굴하지 않았다. She got superior scores on intelligence tests, which helped her secure a scholarship at a prestigious university. 그녀는 지능 검사에서 우수한 점수를 받아 명문 대학 장학금을 받았다. She didn't believe in supernatural things, even though her friends claimed to have experienced ghosts in the old house. 그녀는 친구들이 ...
자세한 내용 보기

[Ubuntu] update-alternatives(Symbolic Link 관리)

자바를 새로 설치시에   /home/tech/jdk7 에 설치했다면 java 실행 파일은 /home/tech/jdk7/bin/java 이다. 기존에 리눅스의 기본 자바인 openJDK 를 새로 설치한 자바로 심볼릭링크 변경을 하려면 아래와 같다 # update-alternatives --list java /usr/lib/jvm/java-6-openjdk-i386/jre/bin/java # update-alternatives --install "/usr/bin/java" "java" "/home/tech/jdk7/bin/java" 1 # update-alternatives --config java 대체 항목 java에 대해 (/usr/bin/java 제공) 2개 선택이 있습니다.   선택       경로                                                                 우선순     상태 ---------------------------------------------------------------------------------------------    0            /usr/lib/jvm/java-6-openjdk-i386/jre/bin/java   1061      자동 모드 * 1            /home/tech...
자세한 내용 보기

언어의 어순과 문화적 특징의 관계를 생각해봅니다.

 "It would be tiresome and very slow as patients recovering from a brain injury affecting the motor system know" 해석은 다음과 같습니다. "운동계에 영향을 미치는 뇌손상에서 회복 중인 환자들이 잘 알고 있듯이, 그 과정은 매우 느리고 또한 피로를 동반하는 것이다." Here are two refined versions in different styles: 1. Formal Academic Version "Such progress would be markedly slow and physically fatiguing, as is well documented among patients recovering from brain injuries that impair the motor system." Uses "markedly slow" and "physically fatiguing" for precision. "As is well documented" signals evidence-based knowledge. "Brain injuries that impair the motor system" is more formal and technical. 2. Plain Everyday Version "It would be really slow and tiring, just like people recovering from a brain injury that affects movement know from experience." Uses "really slow and tiring" for casual tone. Keeps "just like... know from experience...
자세한 내용 보기

how to delete all issues on project in sentry, 센트리 이슈 삭제하기

As of now, Sentry doesn't have a direct feature to delete all issues in bulk through the web interface. However, there are a few methods you can use to archive, resolve, or delete issues programmatically or by adjusting project settings: 1. Bulk Archive Issues from the Web Interface Though there is no mass delete option, you can bulk archive or resolve issues, which essentially hides them from active issue lists. Steps: Go to your Sentry project. In the issue list view, select the issues you want to archive or resolve. Use the "Select All" checkbox to select multiple issues. Choose "Resolve" or "Archive" from the bulk action dropdown. However, this doesn't delete the issues permanently; they will be archived or marked as resolved. 2. Adjust Retention Policy (Data Retention Settings) If you're looking to remove older issues, you can adjust the retention policy at the organization level. Sentry allows you to configure how long issue...
자세한 내용 보기

PySpark Dataframe from HBase

To filter data from HBase and write it back using PySpark, follow these steps: Read data from HBase into a PySpark DataFrame. Filter the DataFrame using PySpark transformations. Write the filtered DataFrame to your desired format (e.g., HBase, Parquet, CSV, etc.). Below is the complete example covering these tasks. 1. Reading, Filtering, and Writing Data Step-by-Step Code from pyspark.sql import SparkSession from pyspark.sql.functions import col # Step 1: Initialize Spark Session with HBase Configurations spark = SparkSession.builder \ .appName( "HBase-PySpark Read-Write" ) \ .config( "spark.jars" , "/path/to/hbase-client.jar,/path/to/hadoop-common.jar" ) \ .config( "spark.hadoop.hbase.zookeeper.quorum" , "localhost" ) \ .config( "spark.hadoop.hbase.zookeeper.property.clientPort" , "2181" ) \ .getOrCreate() # Step 2: Define the HBase Table Catalog hbase_table = "my_table...
자세한 내용 보기