...
Kafka config & Implementation
Example of Existing Consumer (Java)
The below code snippet taken from cps-temporal can be used in the same way in NCMP to listen to message from DMI substituting the topics and errorHandler
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
/** * Consume the specified event. * * @param cpsDataUpdatedEvent the data updated event to be consumed and persisted. */ @KafkaListener(topics = "${app.listener.data-updated.topic}", errorHandler = "dataUpdatedEventListenerErrorHandler") public void consume(final CpsDataUpdatedEvent cpsDataUpdatedEvent) { log.debug("Receiving {} ...", cpsDataUpdatedEvent); // Validate event envelop validateEventEnvelop(cpsDataUpdatedEvent); // Map event to entity final var networkData = this.cpsDataUpdatedEventMapper.eventToEntity(cpsDataUpdatedEvent); log.debug("Persisting {} ...", networkData); // Persist entity final var persistedNetworkData = this.networkDataService.addNetworkData(networkData); log.debug("Persisted {}", persistedNetworkData); } |
Example of Existing Consumer (Config)
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
# ============LICENSE_START======================================================= # Copyright (c) 2021 Bell Canada. # ================================================================================ # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # ============LICENSE_END========================================================= # Spring profile configuration for sasl ssl Kafka spring: kafka: bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVER} security: protocol: SASL_SSL ssl: trust-store-type: JKS trust-store-location: ${KAFKA_SSL_TRUST_STORE_LOCATION} trust-store-password: ${KAFKA_SSL_TRUST_STORE_PASSWORD} properties: sasl.mechanism: SCRAM-SHA-512 sasl.jaas.config: ${KAFKA_SASL_JAAS_CONFIG} ssl.endpoint.identification.algorithm: |
Example of Existing Producer (Java)
Example of Existing Producer (Config)
Example of Kafka Docker-Compose
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
kafka: image: confluentinc/cp-kafka:6.1.1 container_name: kafka ports: - "19092:19092" depends_on: - zookeeper environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,CONNECTIONS_FROM_HOST://localhost:19092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONNECTIONS_FROM_HOST:PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 |
...
Will we be able to handle everything "as-is"?
Future (or alternative)
What are Futures?
CompletableFuture (Java8+)
Java 8 introduced the the CompletableFuture class class. Along with the Future interface, it also implemented the CompletionStage interface. This interface defines the contract for an asynchronous computation step that we can combine with other steps.
CompletableFuture is at the same time a building block and a framework, with about 50 different methods for composing, combining, and executing asynchronous computation steps and handling errors.
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public Future<String> calculateAsync() throws InterruptedException { CompletableFuture<String> completableFuture = new CompletableFuture<>(); Executors.newCachedThreadPool().submit(() -> { Thread.sleep(500); completableFuture.complete("Hello"); return null; }); return completableFuture; } |
...