1.1 On the Instance List page in the CKafka console, click the target instance ID to enter the instance details page.
1.2 In Basic Info > Access Mode, click Add a routing policy. In the pop-up window, select Route Type: Public domain name access, Access Mode: SASL_PLAINTEXT.
2. Create a role.
On the User Management tab page, create a role and set the password.
3. Create a topic.
Create a topic on the Topic Management tab page as instructed in Topic Management.
Step 2. Add the configuration file
1. Add the following dependencies to the pom.xml file:
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.1.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.5</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.6.4</version>
</dependency>
</dependencies>
2. Create a JAAS configuration file named ckafka_client_jaas.conf and modify it with the user created on the User Management tab page.
Set username to a value in the format of instance ID + # + configured username, and password to a configured password.
3. Create a Kafka configuration file named kafka.properties.
## Configure the accessed network by copying the information in the **Network** column in the **Access Mode** section on the instance details** page in the console
bootstrap.servers=ckafka-xxxxxxx
## Configure the topic by copying the information on the **Topic Management** page in the console
// If `ack` is 0, the producer will not wait for acknowledgment from the broker, and the retry configuration will not take effect. If traffic throttling is triggered, the connection will be closed.
// If `ack` is 1, the broker leader will directly return `ack` without waiting for acknowledgment from all broker followers
// If `ack` is `all`, the broker leader will return `ack` only after receiving acknowledgment from all broker followers
props.put(ProducerConfig.ACKS_CONFIG,"all");
//Create a producer object. Note: a producer object is thread-safe, and generally one producer object is sufficient for a process.
//If the sending still fails after client internal retries, the system needs to report and handle the error
System.out.println("error occurred");
}
}
}
2. Compile and run KafkaSaslProducerDemo.java to send messages.
3. View the execution result (output).
Produce ok:ckafka-topic-demo-0@198
Produce ok:ckafka-topic-demo-0@199
4. On the Topic Management tab page on the instance details page in the CKafka console, select the target topic, and click More > Message Query to view the message just sent.
Step 4. Consume messages
1. Create a program named KafkaSaslConsumerDemo.java for a single consumer to subscribe to messages.
//If the consumer does not return a heartbeat message within the interval, the broker will determine that the consumer is not alive, and then remove the consumer from the consumer group and trigger rebalancing. The default value is 30s.
//Set the maximum number of messages that can be polled at a time
//Do not set this parameter to an excessively large value. If polled messages are not all consumed before the next poll starts, load balancing is triggered and lagging occurs.
ConsumerRecords<String, String> records = consumer.poll(1000);
//All messages must be consumed before the next poll, and the total duration cannot exceed the timeout interval specified by `SESSION_TIMEOUT_MS_CONFIG`
2. Compile and run KafkaSaslConsumerDemo.java to consume messages.
3. View the execution result (output).
Consume partition:0 offset:298
Consume partition:0 offset:299
4. On the Consumer Group page in the Ckafka console, select the consumer group name, enter the topic name, and click Query Details to view the consumption details.
Yes
No
Was this page helpful?