Developing Event-driven Applications with Apache Kafka and Red Hat AMQ Streams

Procedure 2.5. Instructions

1. In your workspace, move to the kafka-basics directory. Create a topic that conforms to the following specifications:

   Table 2.1. Topic Specifications

   Name	wind-turbine-production
   Partitions	2
   Retention time (ms)	604800000
   Segment bytes	1073741824
   Cluster	my-cluster

   
   

   1. By using your editor of choice, open the resources/application-topic.yaml file, and use the Red Hat AMQ Streams topic operator to create the topic.

      The file content should look like the following:

      apiVersion: kafka.strimzi.io/v1beta2
      kind: KafkaTopic
      metadata:
        name: wind-turbine-production
        labels:
          strimzi.io/cluster: my-cluster
      spec:
        partitions: 2
        replicas: 1
        config:
          retention.ms: 604800000
          segment.bytes: 1073741824

   2. Verify that your project is set to RHT_OCP4_DEV_USER-kafka-cluster.

      (.venv) [user@host kafka-basics]$ oc project
      Using project "RHT_OCP4_DEV_USER-kafka-cluster" on server ...output omitted...

   3. Use the oc command to create the topic in the Kafka cluster project.

      (.venv) [user@host kafka-basics]$ oc create -f resources/application-topic.yaml
      kafkatopic.kafka.strimzi.io/wind-turbine-production created

   Hide Solution

2. Move to the energy-meter directory, which contains the source code of the application. Configure the Kafka connection security settings in the Config class. Use the following configuration properties to connect to the Kafka cluster:

   • Security protocol: SSL

   • Location of your TrustStore file

   • Password: password

   1. Open the Config class and configure the connection protocol to use SSL in the configureConnectionSecurity method.

      // TODO: configure the connection protocol
      props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");

   2. Set the absolute path to the TrustStore file.

      // TODO: configure the path to the truststore file
      props.put(
          SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
          "ABSOLUTE_PATH_TO_YOUR_WORKSPACE_FOLDER/truststore.jks"
      );

   3. Set the TrustStore password.

      // TODO: configure the truststore password
      props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "password");

   Hide Solution

3. Implement the business logic of the WindTurbine class. Given a sequence of electricity production values stored in an attribute named energyProductionSequence, create the business logic to send each one of the values to the wind-turbine-production topic. Send each integer record without a key.

   Configure the WindTurbine class as a producer that conforms to the following configuration:

   Table 2.2. Producer Configuration

   Key serializer	StringSerializer
   Value serializer	IntegerSerializer

   
   

   You can use the printRecord helper method for debugging purposes.

   1. Open the WindTurbine class and examine the code. You can use the helper methods for debugging purposes.

   2. Set the bootstrap server in the configureProducer method.

      // TODO: configure the bootstrap server
      props.put(
          ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
          "YOUR_KAFKA_BOOTSTRAP_HOST:YOUR_KAFKA_BOOTSTRAP_PORT"
      );

   3. Set the key and value serializers in the configureProducer method.

      // TODO: configure the key serializer
      props.put(
          ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringSerializer"
      );
      
      // TODO: configure the value serializer
      props.put(
          ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.IntegerSerializer"
      );

   4. Implement the business logic in the main method to send the energy production values to the wind-turbine-production topic. Do not assign a key to the records.

      The method should look like the following:

      public static void main(String[] args) {
          // TODO: implement the business logic
          Producer<Void,Integer> producer = new KafkaProducer<>(
                  configureProperties()
          );
      
          for (int energyProduction : energyProductionSequence) {
              ProducerRecord<Void, Integer> record = new ProducerRecord<>(
                      "wind-turbine-production",
                      energyProduction
              );
      
              producer.send(record);
              printRecord(record);
          }
      
          producer.close();
      }

   5. Execute the application to send the electricity production values to the wind-turbine-production topic.

      (.venv) [user@host energy-meter]$ ./mvnw compile exec:java \
       "-Dexec.mainClass=com.redhat.energy.meter.producer.WindTurbine"
      ...output omitted...
      Sent record:
              Topic = wind-turbine-production
              Partition = null
              Key = null
              Value = 700
      ...output omitted...

   Hide Solution

4. Implement the business logic of the ReportingSystem class. This class consumes the measures available in the wind-turbine-production topic, sums the values of every 5 processed records, and stores the value in a file named report.txt.

   Configure the ReportingSystem class as a consumer that conforms to the following configuration:

   Table 2.3. Consumer Configuration

   Group ID	reportingSystem
   Key deserializer	StringDeserializer
   Value deserializer	IntegerDeserializer
   Offset reset config	earliest

   
   

   You can use the print helper methods for debugging purposes.

   1. Open the ReportingSystem class and examine the code. You can use the helper methods for debugging purposes.

   2. Set the bootstrap server in the configureConsumer method.

      // TODO: set the bootstrap server
      props.put(
          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
          "YOUR_KAFKA_BOOTSTRAP_HOST:YOUR_KAFKA_BOOTSTRAP_PORT"
      );

   3. Set the consumer group ID in the configureConsumer method.

      // TODO: set the consumer group ID
      props.put(
          ConsumerConfig.GROUP_ID_CONFIG,
          "reportingSystem"
      );

   4. Set the key and value deserializers in the configureConsumer method.

      // TODO: set the key deserializer
      props.put(
          ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringDeserializer"
      );
      
      // TODO: set the value deserializer
      props.put(
          ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.IntegerDeserializer"
      );

   5. Set the offset reset configuration to earliest in the configureConsumer method.

      // TODO: set the offset reset config
      props.put(
          ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
          "earliest"
      );

   6. Implement the business logic in the main method to consume the measures available in the wind-turbine-production topic, sum the values of every 5 processed records, and store the value in a file named report.txt. You must use the saveAggregationToFile helper method to save the sum value.

      The method should look like the following:

      public static void main( String[] args ) throws IOException {
          // TODO: implement the business logic
          Consumer<Void,Integer> consumer = new KafkaConsumer<>(configureProperties());
          consumer.subscribe(Collections.singletonList("wind-turbine-production"));
      
          while (true) {
              ConsumerRecords<Void, Integer> records = consumer.poll(Duration.ofSeconds(10));
      
              int aggregatedEnergy = 0;
              int processedRecords = 0;
      
              for (ConsumerRecord<Void, Integer> record : records) {
                  printRecord(record);
      
                  aggregatedEnergy += record.value();
                  processedRecords++;
      
                  if (processedRecords % 5 == 0) {
                      printAggregation(aggregatedEnergy);
                      saveAggregationToFile(aggregatedEnergy);
                      aggregatedEnergy = 0;
                  }
              }
          }
      }

   7. Execute the application to process the electricity production values available in the wind-turbine-production topic.

      (.venv) [user@host energy-meter]$ ./mvnw compile exec:java \
       "-Dexec.mainClass=com.redhat.energy.meter.consumer.ReportingSystem"
      ...output omitted...
      Received record:
              Topic = wind-turbine-production
              Partition = 1
              Key = null
              Value = 700
      Writing aggregation result to file: 2500

   8. Wait until the consumer processes 5 records and writes the sum to the report.txt file. Press CTRL + c to stop the application.

   Hide Solution