### Implementing SourceConnector.start() and stop() Source: https://github.com/apache/kafka/blob/trunk/docs/kafka-connect/connector-development-guide.md Implement the `start()` method for initialization logic and resource setup. The `stop()` method is used for cleanup when the connector is no longer needed. ```java @Override public void start(Map props) { // Initialization logic and setting up of resources can take place in this method. // This connector doesn't need to do any of that, but we do log a helpful message to the user. this.props = props; AbstractConfig config = new AbstractConfig(CONFIG_DEF, props); String filename = config.getString(FILE_CONFIG); filename = (filename == null || filename.isEmpty()) ? "standard input" : config.getString(FILE_CONFIG); log.info("Starting file source connector reading from {}", filename); } @Override public void stop() { // Nothing to do since no background monitoring is required. } ``` -------------------------------- ### Start Kafka Cluster Source: https://github.com/apache/kafka/blob/trunk/trogdor/README.md Initialize and start a single-node Kafka cluster for testing. ```bash KAFKA_CLUSTER_ID="$(./bin/kafka-storage.sh random-uuid)" ./bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties ./bin/kafka-server-start.sh config/server.properties &> /tmp/kafka.log & ``` -------------------------------- ### Start Kafka Server Source: https://github.com/apache/kafka/blob/trunk/docs/getting-started/quickstart.md Start the Kafka server using the configuration file. This command launches a basic Kafka environment ready for use. ```bash bin/kafka-server-start.sh config/server.properties ``` -------------------------------- ### Start Kafka Containers with Docker Compose Source: https://github.com/apache/kafka/blob/trunk/docker/examples/README.md Use this command to bring up Kafka containers using the apache/kafka Docker image. Ensure you have Docker Compose installed and configured. ```bash IMAGE=apache/kafka:latest ``` -------------------------------- ### Start Kafka Streams Application Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/running-app.md Package your Java application as a fat JAR and start it using the `java -cp` command, specifying the JAR path and the main class. ```bash # Start the application in class `com.example.MyStreamsApp` # from the fat JAR named `path-to-app-fatjar.jar`. $ java -cp path-to-app-fatjar.jar com.example.MyStreamsApp ``` -------------------------------- ### Example Server Certificate Details Source: https://github.com/apache/kafka/blob/trunk/docs/security/encryption-and-authentication-using-ssl.md This is an example of the server's certificate details that should appear in the openssl s_client output if the keystore is configured properly. ```text -----BEGIN CERTIFICATE----- {variable sized random bytes} -----END CERTIFICATE----- subject=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=Sriharsha Chintalapani issuer=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=kafka/emailAddress=test@test.com ``` -------------------------------- ### Start WordCount Demo with KIP-1071 Protocol Source: https://github.com/apache/kafka/blob/trunk/docs/streams/quickstart.md Starts the WordCount demo application using the new rebalancing protocol (KIP-1071) for improved rebalance times. Requires a properties file. ```bash $ echo "group.protocol=streams" > streams.properties $ bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo streams.properties ``` -------------------------------- ### Example Custom Provider Configuration Source: https://github.com/apache/kafka/blob/trunk/docs/configuration/configuration-providers.md Illustrates how to configure a custom configuration provider, including its class and parameters. ```properties config.providers=customProvider config.providers.customProvider.class=com.example.customProvider config.providers.customProvider.param.param1=value1 config.providers.customProvider.param.param2=value2 ``` -------------------------------- ### Setup Python Virtual Environment and Install Dependencies Source: https://github.com/apache/kafka/blob/trunk/release/README.md Creates a Python virtual environment named '.venv', activates it, and installs project dependencies from 'requirements.txt'. This ensures a clean and isolated environment for the release process. ```bash python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt ``` -------------------------------- ### Setup Kafka Streams Topology for Testing Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/testing.md Demonstrates the setup of a Kafka Streams topology, including sources, processors, state stores, and sinks, within a test environment. It configures the TopologyTestDriver and creates test input/output topics. The state store is pre-populated for testing. ```java private TopologyTestDriver testDriver; private TestInputTopic inputTopic; private TestOutputTopic outputTopic; private KeyValueStore store; private Serde stringSerde = new Serdes.StringSerde(); private Serde longSerde = new Serdes.LongSerde(); @Before public void setup() { Topology topology = new Topology(); topology.addSource("sourceProcessor", "input-topic"); topology.addProcessor("aggregator", new CustomMaxAggregatorSupplier(), "sourceProcessor"); topology.addStateStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore("aggStore"), Serdes.String(), Serdes.Long()).withLoggingDisabled(), // need to disable logging to allow store pre-populating "aggregator"); topology.addSink("sinkProcessor", "result-topic", "aggregator"); // setup test driver Properties props = new Properties(); props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass().getName()); testDriver = new TopologyTestDriver(topology, props); // setup test topics inputTopic = testDriver.createInputTopic("input-topic", stringSerde.serializer(), longSerde.serializer()); outputTopic = testDriver.createOutputTopic("result-topic", stringSerde.deserializer(), longSerde.deserializer()); // pre-populate store store = testDriver.getKeyValueStore("aggStore"); store.put("a", 21L); } @After public void tearDown() { testDriver.close(); } ``` -------------------------------- ### Run a Kafka broker Source: https://github.com/apache/kafka/blob/trunk/README.md Commands to start a broker using compiled files or a Docker image. ```bash KAFKA_CLUSTER_ID="$(./bin/kafka-storage.sh random-uuid)" ./bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties ./bin/kafka-server-start.sh config/server.properties ``` ```bash docker run -p 9092:9092 apache/kafka:latest ``` -------------------------------- ### Task Specification Examples Source: https://github.com/apache/kafka/blob/trunk/trogdor/README.md JSON configurations for defining network partition faults and produce benchmark workloads. ```json { "class": "org.apache.kafka.trogdor.fault.NetworkPartitionFaultSpec", "startMs": 1000, "durationMs": 30000, "partitions": [["node1", "node2"], ["node3"]] } ``` ```json { "class": "org.apache.kafka.trogdor.workload.ProduceBenchSpec", "durationMs": 10000000, "producerNode": "node0", "bootstrapServers": "localhost:9092", "targetMessagesPerSec": 10000, "maxMessages": 50000, "activeTopics": { "foo[1-3]": { "numPartitions": 10, "replicationFactor": 1 } }, "inactiveTopics": { "foo[4-5]": { "numPartitions": 10, "replicationFactor": 1 } }, "keyGenerator": { "type": "sequential", "size": 8, "offset": 1 }, "useConfiguredPartitioner": true } ``` -------------------------------- ### Example SSL Principal Mapping Rules Source: https://github.com/apache/kafka/blob/trunk/docs/security/authorization-and-acls.md Provides examples of `ssl.principal.mapping.rules` for translating distinguished names to short names, including case conversion. ```text RULE:^CN=(.*?),OU=ServiceUsers.*$/$1/, RULE:^CN=(.*?),OU=(.*?),O=(.*?),L=(.*?),ST=(.*?),C=(.*?)$/$1@$2/L, RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L, DEFAULT ``` -------------------------------- ### Start Console Producer Source: https://github.com/apache/kafka/blob/trunk/docs/streams/quickstart.md Starts a console producer to send messages to a specified Kafka topic. Use this to input data for the demo application. ```bash $ bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic streams-plaintext-input ``` -------------------------------- ### Start KRaft Controller Source: https://github.com/apache/kafka/blob/trunk/docs/operations/hardware-and-os.md Start the KRaft controller after successfully formatting the metadata log directories. This command resumes controller operations with the new disk. ```bash $ bin/kafka-server-start.sh config/server.properties ``` -------------------------------- ### Publish streams quickstart to Maven Source: https://github.com/apache/kafka/blob/trunk/README.md Deploys the streams archetype artifact to a Maven repository. ```bash cd streams/quickstart mvn deploy ``` -------------------------------- ### Start MirrorMaker Process Source: https://github.com/apache/kafka/blob/trunk/docs/operations/geo-replication-(cross-cluster-data-mirroring).md Initiates a MirrorMaker process using a specified properties file. This is the basic command to begin cross-cluster data replication. ```bash $ bin/connect-mirror-maker.sh connect-mirror-maker.properties ``` -------------------------------- ### Start Console Consumer for Output Topic Source: https://github.com/apache/kafka/blob/trunk/docs/streams/quickstart.md Starts a console consumer to read messages from the output topic. Configured to print keys and values with specific deserializers. ```bash $ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic streams-wordcount-output \ --from-beginning \ --formatter-property print.key=true \ --formatter-property print.value=true \ --formatter-property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \ --formatter-property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer ``` -------------------------------- ### Start WordCount Demo Application Source: https://github.com/apache/kafka/blob/trunk/docs/streams/quickstart.md Launches the WordCount demo application. This application processes messages from an input topic and writes results to an output topic. ```bash $ bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo ``` -------------------------------- ### Download and Extract Kafka Source: https://github.com/apache/kafka/blob/trunk/docs/getting-started/quickstart.md Download the latest Kafka release and extract it to your local machine. Navigate into the extracted directory to begin setup. ```bash tar -xzf kafka_2.13-4.3.0.tgz cd kafka_2.13-4.3.0 ``` -------------------------------- ### Describe Kafka Topics Source: https://github.com/apache/kafka/blob/trunk/docs/streams/quickstart.md Describes the configuration and partition details of the created Kafka topics. Useful for verifying topic setup. ```bash $ bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --exclude-internal Topic:streams-wordcount-output PartitionCount:1 ReplicationFactor:1 Configs:cleanup.policy=compact,segment.bytes=1073741824 Topic: streams-wordcount-output Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic:streams-plaintext-input PartitionCount:1 ReplicationFactor:1 Configs:segment.bytes=1073741824 Topic: streams-plaintext-input Partition: 0 Leader: 0 Replicas: 0 Isr: 0 ``` -------------------------------- ### Implementing SourceConnector.taskConfigs() Source: https://github.com/apache/kafka/blob/trunk/docs/kafka-connect/connector-development-guide.md Implement `taskConfigs()` to determine the number of tasks and their configurations. This example returns a single configuration map for a single input file. ```java @Override public List> taskConfigs(int maxTasks) { // Note that the task configs could contain configs additional to or different from the connector configs if needed. For instance, // if different tasks have different responsibilities, or if different tasks are meant to process different subsets of the source data stream). ArrayList> configs = new ArrayList<>(); // Only one input stream makes sense. configs.add(props); return configs; } ``` -------------------------------- ### Kafka Console Consumer with SSL Source: https://github.com/apache/kafka/blob/trunk/docs/security/encryption-and-authentication-using-ssl.md Example of how to use the Kafka console consumer with SSL-enabled properties. ```bash $ bin/kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --command-config client-ssl.properties ``` -------------------------------- ### Run Trogdor Daemons Source: https://github.com/apache/kafka/blob/trunk/trogdor/README.md Start the Trogdor agent and coordinator processes. ```bash ./bin/trogdor.sh agent -c ./config/trogdor.conf -n node0 &> /tmp/trogdor-agent.log & ``` ```bash ./bin/trogdor.sh coordinator -c ./config/trogdor.conf -n node0 &> /tmp/trogdor-coordinator.log & ``` -------------------------------- ### Example of Keyed Messages for Log Compaction Source: https://github.com/apache/kafka/blob/trunk/docs/design/design.md This example illustrates a stream of messages for a single user ID where each message represents an update to their email address. Log compaction ensures that only the latest email address for a given user ID is retained. ```text 123 => bill@microsoft.com . . . 123 => bill@gatesfoundation.org . . . 123 => bill@gmail.com ``` -------------------------------- ### Kafka Console Producer with SSL Source: https://github.com/apache/kafka/blob/trunk/docs/security/encryption-and-authentication-using-ssl.md Example of how to use the Kafka console producer with SSL-enabled properties. ```bash $ bin/kafka-console-producer.sh --bootstrap-server localhost:9093 --topic test --command-config client-ssl.properties ``` -------------------------------- ### Custom Deserialization Exception Handler Example Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/config-streams.md Implement a custom deserialization exception handler to forward corrupt records to a dead-letter queue. This example shows how to create a KafkaProducer and send records to a specified topic within the handler. ```java public class SendToDeadLetterQueueExceptionHandler implements DeserializationExceptionHandler { KafkaProducer dlqProducer; String dlqTopic; @Override public DeserializationHandlerResponse handle(final ErrorHandlerContext context, final ConsumerRecord record, final Exception exception) { log.warn("Exception caught during Deserialization, sending to the dead queue topic; " + "taskId: {}, topic: {}, partition: {}, offset: {}", context.taskId(), record.topic(), record.partition(), record.offset(), exception); dlqProducer.send(new ProducerRecord<>(dlqTopic, record.timestamp(), record.key(), record.value(), record.headers())).get(); return DeserializationHandlerResponse.CONTINUE; } @Override public void configure(final Map configs) { dlqProducer = .. // get a producer from the configs map dlqTopic = .. // get the topic name from the configs map } } ``` -------------------------------- ### Run Single KRaft Quorum Source: https://github.com/apache/kafka/blob/trunk/raft/README.md Starts a single KRaft quorum using a specified properties file and replica directory ID. This is a basic setup for testing. ```bash bin/test-kraft-server-start.sh --config config/kraft.properties --replica-directory-id b8tRS7h4TJ2Vt43Dp85v2A ``` -------------------------------- ### Kafka Streams Security Configuration Error Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/security.md This example shows a typical runtime error when a security setting, such as `ssl.keystore.password`, is misconfigured. Monitor application logs for such exceptions to quickly identify and resolve security setup issues. ```text # Misconfigured ssl.keystore.password Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to construct kafka producer [...snip...] Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: java.io.IOException: Keystore was tampered with, or password was incorrect [...snip...] Caused by: java.security.UnrecoverableKeyException: Password verification failed ``` -------------------------------- ### Set Up Python Virtual Environment Source: https://github.com/apache/kafka/blob/trunk/committer-tools/README.md Provides commands to create and activate a Python virtual environment. This is an optional but recommended step to isolate project dependencies. ```bash python -m venv venv # For Linux/macOS source venv/bin/activate # On Windows: # .\venv\Scripts\activate ``` -------------------------------- ### Produce Messages to Single Node Kafka Source: https://github.com/apache/kafka/blob/trunk/docker/examples/README.md Use this command to produce messages to a Kafka topic using the console producer. Ensure Java version 17 or higher is installed. This command is for a single-node setup with SASL Plaintext. ```bash # Run from root of the repo $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9094 --command-config ./docker/examples/fixtures/client-secrets/client-sasl.properties ``` -------------------------------- ### Prepare for Exactly-Once Source Support Upgrade Source: https://github.com/apache/kafka/blob/trunk/docs/operations/geo-replication-(cross-cluster-data-mirroring).md For existing MirrorMaker clusters, set `exactly.once.source.support` to `preparing` as the first step in a two-step upgrade process. ```properties us-east.exactly.once.source.support = preparing ``` -------------------------------- ### Configure and Run Multi-Node KRaft Quorum Source: https://github.com/apache/kafka/blob/trunk/raft/README.md Sets up and starts a multi-node KRaft quorum by defining properties for three separate nodes and then launching each node. This configuration is for a more robust testing environment. ```bash cat << EOF >> config/kraft-quorum-1.properties node.id=1 listeners=PLAINTEXT://localhost:9092 controller.listener.names=PLAINTEXT controller.quorum.voters=1@localhost:9092,2@localhost:9093,3@localhost:9094 log.dirs=/tmp/kraft-logs-1 EOF cat << EOF >> config/kraft-quorum-2.properties node.id=2 listeners=PLAINTEXT://localhost:9093 controller.listener.names=PLAINTEXT controller.quorum.voters=1@localhost:9092,2@localhost:9093,3@localhost:9094 log.dirs=/tmp/kraft-logs-2 EOF cat << EOF >> config/kraft-quorum-3.properties node.id=3 listeners=PLAINTEXT://localhost:9094 controller.listener.names=PLAINTEXT controller.quorum.voters=1@localhost:9092,2@localhost:9093,3@localhost:9094 log.dirs=/tmp/kraft-logs-3 EOF bin/test-kraft-server-start.sh --config config/kraft-quorum-1.properties --replica-directory-id b8tRS7h4TJ2Vt43Dp85v2A bin/test-kraft-server-start.sh --config config/kraft-quorum-2.properties --replica-directory-id Nkij_D9XRiYKNb41SiJo7Q bin/test-kraft-server-start.sh --config config/kraft-quorum-3.properties --replica-directory-id 4-e97nI7eHPYKfEDtW8rtQ ``` -------------------------------- ### WordCount Application in Scala Source: https://github.com/apache/kafka/blob/trunk/docs/streams/introduction.md Implements a WordCount application using Kafka Streams in Scala. This example mirrors the Java version, demonstrating stream processing, word counting, and outputting results. It includes setup for Kafka Streams configuration and utilizes Scala's implicit conversions for a more concise syntax. ```scala import java.util.Properties import java.util.concurrent.TimeUnit import org.apache.kafka.streams.kstream.Materialized import org.apache.kafka.streams.scala.ImplicitConversions._ import org.apache.kafka.streams.scala._ import org.apache.kafka.streams.scala.kstream._ import org.apache.kafka.streams.{KafkaStreams, StreamsConfig} object WordCountApplication extends App { import Serdes._ val props: Properties = { val p = new Properties() p.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application") p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092") p } val builder: StreamsBuilder = new StreamsBuilder val textLines: KStream[String, String] = builder.stream[String, String]("TextLinesTopic") val wordCounts: KTable[String, Long] = textLines .flatMapValues(textLine => textLine.toLowerCase.split("\\W+")) .groupBy((_, word) => word) .count()(Materialized.as("counts-store")) wordCounts.toStream.to("WordsWithCountsTopic") val streams: KafkaStreams = new KafkaStreams(builder.build(), props) streams.start() sys.ShutdownHookThread { streams.close(10, TimeUnit.SECONDS) } } ``` -------------------------------- ### Describe Kafka Share Group Offsets Source: https://github.com/apache/kafka/blob/trunk/docs/operations/basic-kafka-operations.md View the current start offset and lag for a specific share group and topic. The admin client requires DESCRIBE access to all topics used by the group. ```bash $ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group my-share-group GROUP TOPIC PARTITION START-OFFSET LAG my-share-group topic1 0 4 0 ``` -------------------------------- ### Another Static KRaft Quorum Example Source: https://github.com/apache/kafka/blob/trunk/docs/operations/kraft.md A variation of the static KRaft quorum example, showing a different `metadata.version`. ```text Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.8-IV0 FinalizedVersionLevel: 3.8-IV0 Epoch: 5 ``` -------------------------------- ### SourceTask Lifecycle Methods Source: https://github.com/apache/kafka/blob/trunk/docs/kafka-connect/connector-development-guide.md Implements the `start` and `stop` lifecycle methods for a Kafka Connect SourceTask. The `start` method initializes task properties from configuration, and `stop` handles resource cleanup. Note that `start` does not yet handle resuming from previous offsets. ```java public class FileStreamSourceTask extends SourceTask { private String filename; private InputStream stream; private String topic; private int batchSize; @Override public void start(Map props) { filename = props.get(FileStreamSourceConnector.FILE_CONFIG); stream = openOrThrowError(filename); topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG); batchSize = props.get(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG); } @Override public synchronized void stop() { stream.close(); } } ``` -------------------------------- ### Custom RocksDB Configuration Setter Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/config-streams.md Implement RocksDBConfigSetter to provide custom RocksDB configurations. This example adjusts memory size, block size, caching, and write buffer settings. Ensure to close any opened RocksDB objects in the close method to prevent memory leaks. ```java public static class CustomRocksDBConfig implements RocksDBConfigSetter { // This object should be a member variable so it can be closed in RocksDBConfigSetter#close. private org.rocksdb.Cache cache = new org.rocksdb.LRUCache(16 * 1024L * 1024L); @Override public void setConfig(final String storeName, final Options options, final Map configs) { // See #1 below. BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig(); tableConfig.setBlockCache(cache); // See #2 below. tableConfig.setBlockSize(16 * 1024L); // See #3 below. tableConfig.setCacheIndexAndFilterBlocks(true); options.setTableFormatConfig(tableConfig); // See #4 below. options.setMaxWriteBufferNumber(2); } @Override public void close(final String storeName, final Options options) { // See #5 below. cache.close(); } } ``` ```java Properties streamsSettings = new Properties(); streamsConfig.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, CustomRocksDBConfig.class); ``` -------------------------------- ### Dynamic KRaft Quorum Example Source: https://github.com/apache/kafka/blob/trunk/docs/operations/kraft.md An example output indicating a dynamic KRaft quorum, where `kraft.version` is finalized at level 1 or above. ```text Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 1 Epoch: 5 Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.9-IV0 FinalizedVersionLevel: 3.9-IV0 Epoch: 5 ``` -------------------------------- ### Verify SSL Setup with openssl Source: https://github.com/apache/kafka/blob/trunk/docs/security/encryption-and-authentication-using-ssl.md Use the openssl command-line tool to quickly check if the server keystore and truststore are set up correctly. This command connects to the specified SSL port and attempts TLSv1. ```bash $ openssl s_client -debug -connect localhost:9093 -tls1 ``` -------------------------------- ### Static KRaft Quorum Example Source: https://github.com/apache/kafka/blob/trunk/docs/operations/kraft.md An example output indicating a static KRaft quorum, where `kraft.version` is finalized at level 0. ```text Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 0 Epoch: 5 Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.9-IV0 FinalizedVersionLevel: 3.9-IV0 Epoch: 5 ``` -------------------------------- ### Implement Custom Production Exception Handler Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/config-streams.md Implement a custom ProductionExceptionHandler to manage specific exceptions like RecordTooLargeException. This example shows how to ignore records that are too large and fail on other exceptions. Ensure to import necessary classes. ```java import java.util.Properties; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.common.errors.RecordTooLargeException; import org.apache.kafka.streams.errors.ProductionExceptionHandler; import org.apache.kafka.streams.errors.ProductionExceptionHandler.ProductionExceptionHandlerResponse; public class IgnoreRecordTooLargeHandler implements ProductionExceptionHandler { public void configure(Map config) {} public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context, final ProducerRecord record, final Exception exception) { if (exception instanceof RecordTooLargeException) { return ProductionExceptionHandlerResponse.CONTINUE; } else { return ProductionExceptionHandlerResponse.FAIL; } } } ``` ```properties Properties settings = new Properties(); // other various kafka streams settings, e.g. bootstrap servers, application id, etc settings.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, IgnoreRecordTooLargeHandler.class); ``` -------------------------------- ### Handling Transactional Exceptions Source: https://github.com/apache/kafka/blob/trunk/docs/design/design.md Example code demonstrating how to handle different categories of exceptions that can occur during Kafka transactional operations. Applications must implement specific recovery strategies for `AbortableException`, `ApplicationRecoverableException`, and `InvalidConfigurationException`. ```java try { // Transactional operations } catch (RetriableException e) { // Client retries automatically } catch (RefreshRetriableException e) { // Client refreshes metadata and retries automatically } catch (AbortableException e) { // Abort transaction and reset consumer position producer.abortTransaction(); consumer.seekToBeginning(consumer.assignment()); } catch (ApplicationRecoverableException e) { // Application-specific recovery, may include restarting producer } catch (InvalidConfigurationException e) { // Application-specific handling, may choose to restart producer } catch (KafkaException e) { // General Kafka exception handling } ``` -------------------------------- ### Describe Share Group State Source: https://github.com/apache/kafka/blob/trunk/docs/operations/basic-kafka-operations.md Get a summary of the share group's state, including the coordinator, current state (e.g., Stable), and the number of active members. ```bash $ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group my-share-group --state GROUP COORDINATOR (ID) STATE #MEMBERS my-share-group localhost:9092 (1) Stable 2 ``` -------------------------------- ### Inject ClusterInstance into Test Constructor for Setup (Java) Source: https://github.com/apache/kafka/blob/trunk/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/README.md Shows how to inject ClusterInstance into the test class constructor for common setup logic, such as creating topics before each test. This promotes code reuse for shared setup procedures. ```java class SampleTest { private final ClusterInstance cluster; SampleTest(ClusterInstance cluster) { this.cluster = cluster; } @BeforeEach public void setup() { // Common setup code with started ClusterInstance this.cluster.admin().createTopics(...); } @ClusterTest public void testOne() { // Test code } } ``` -------------------------------- ### Initialize Processor with Mock Context and Properties Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/testing.md Initialize a processor with a MockProcessorContext, providing configuration properties and default Serdes. ```java final Properties props = new Properties(); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass()); props.put("some.other.config", "some config value"); final MockProcessorContext context = new MockProcessorContext<>(props); ``` -------------------------------- ### Example AI-Generated Contribution Commit Source: https://github.com/apache/kafka/blob/trunk/CONTRIBUTING.md An example commit message demonstrating the use of AI-generated contribution trailers for a Kafka ticket. ```git commit KAFKA-9999: adding AGENTS.md Co-Authored-By: Claude Sonnet 4.6 ``` -------------------------------- ### Create Initial SCRAM Credentials Source: https://github.com/apache/kafka/blob/trunk/docs/security/authentication-using-sasl.md Use kafka-storage.sh to format storage and add initial SCRAM credentials for a user. This is typically done before starting Kafka brokers. ```bash $ bin/kafka-storage.sh format -t $(bin/kafka-storage.sh random-uuid) \ -c config/server.properties \ --add-scram 'SCRAM-SHA-256=[name="admin",password="admin-secret"]' ``` -------------------------------- ### Start Streams Application with Shutdown Hook Source: https://github.com/apache/kafka/blob/trunk/docs/streams/tutorial.md Start the Kafka Streams application and set up a shutdown hook to gracefully close the client on termination. ```java final CountDownLatch latch = new CountDownLatch(1); // attach shutdown handler to catch control-c Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") { @Override public void run() { streams.close(); latch.countDown(); } }); try { streams.start(); latch.await(); } catch (Throwable e) { System.exit(1); } System.exit(0); ``` -------------------------------- ### Build JAR and Source JAR Source: https://github.com/apache/kafka/blob/trunk/README.md Commands to compile the project into a standard JAR or a source JAR. ```bash ./gradlew jar ``` ```bash ./gradlew srcJar ``` -------------------------------- ### Describe Cluster-Wide Default Configurations Source: https://github.com/apache/kafka/blob/trunk/docs/configuration/broker-configs.md Use this command to view the currently configured dynamic cluster-wide default settings. This is helpful for understanding the baseline configuration for all brokers. ```bash $ bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-default --describe ``` -------------------------------- ### Create Seed Data for Kafka Connect Source: https://github.com/apache/kafka/blob/trunk/docs/getting-started/quickstart.md Create a test file with seed data for Kafka Connect to import. The command differs slightly for Windows. ```bash $ echo -e "foo bar" > test.txt ``` ```bash $ echo foo > test.txt $ echo bar >> test.txt ``` -------------------------------- ### Run File Input Kafka (Native) Source: https://github.com/apache/kafka/blob/trunk/docker/examples/README.md Launches a single-node Kafka instance where SSL configurations are provided via file input using the GraalVM-based Native Docker image. Ensure you run from the repository root. ```bash IMAGE=apache/kafka-native:latest docker compose -f docker/examples/docker-compose-files/single-node/file-input/docker-compose.yml up ``` -------------------------------- ### Start Kafka Streams Processing Source: https://github.com/apache/kafka/blob/trunk/docs/streams/developer-guide/write-streams-app.md Initiate the Kafka Streams processing by calling the start() method on the KafkaStreams instance. This begins the task of processing data from input topics. ```java // Start the Kafka Streams threads streams.start(); ``` -------------------------------- ### Create Topic with Custom Configurations Source: https://github.com/apache/kafka/blob/trunk/docs/configuration/topic-configs.md Use this command to create a new topic with specific configuration overrides for message size and flush rate. Ensure the topic name, partitions, and replication factor are set as needed. ```bash bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 1 \ --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1 ``` -------------------------------- ### Start Isolated SSL Kafka Cluster (JVM) Source: https://github.com/apache/kafka/blob/trunk/docker/examples/README.md Starts an isolated Kafka cluster with SSL encryption for inter-broker and client communication using the JVM-based Docker image. ```bash # Run from root of the repo # JVM based Apache Kafka Docker Image $ IMAGE=apache/kafka:latest docker compose -f docker/examples/docker-compose-files/cluster/isolated/ssl/docker-compose.yml up ``` -------------------------------- ### Start MirrorMaker with Specific Cluster Target Source: https://github.com/apache/kafka/blob/trunk/docs/operations/geo-replication-(cross-cluster-data-mirroring).md Starts a MirrorMaker process and restricts data replication to a specific cluster alias defined in the configuration file. This is useful for targeted replication. ```bash # Note: The cluster alias us-west must be defined in the configuration file $ bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters us-west ``` -------------------------------- ### List Available Plugins for Migration Source: https://github.com/apache/kafka/blob/trunk/docs/kafka-connect/user-guide.md Use the 'list' subcommand with the Kafka Connect migration script to view plugins available for migration. Specify plugin locations using --worker-config, --plugin-path, or --plugin-location. ```bash bin/connect-plugin-path.sh list --plugin-path ``` -------------------------------- ### Client JAAS Static Configuration Example Source: https://github.com/apache/kafka/blob/trunk/docs/security/authentication-using-sasl.md Configure SASL authentication for Kafka clients using a static JAAS configuration file. This example shows GSSAPI (Kerberos) configuration with a keytab. ```text KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/etc/security/keytabs/kafka_client.keytab" principal="kafka-client-1@EXAMPLE.COM"; }; ``` -------------------------------- ### Run File Input Kafka (JVM) Source: https://github.com/apache/kafka/blob/trunk/docker/examples/README.md Launches a single-node Kafka instance where SSL configurations are provided via file input using the JVM-based Docker image. Ensure you run from the repository root. ```bash IMAGE=apache/kafka:latest docker compose -f docker/examples/docker-compose-files/single-node/file-input/docker-compose.yml up ``` -------------------------------- ### Check Python and Pip Installation Source: https://github.com/apache/kafka/blob/trunk/committer-tools/README.md Verifies if Python 3.x and pip are installed on the system by checking their versions. This is a prerequisite for using most of the committer tools. ```bash python --version pip --version ```