Showing posts with label Spring. Show all posts
Showing posts with label Spring. Show all posts

Friday, June 5, 2020

Auditing in MySQL

One of the important requirements in many RDBMS based workloads is to have audit log where any row in any table is stamped with who changed it and  when was it changed? To do this in every place where the table is modified is extremely cumbersome. In this blog post we look at how we can enable the springframework based auditing framework to perform this activity.

Before we do that, we need to upgrade the versions for our dependencies since I have been using this project for quite sometime. Also I decided to use lombok logging annotation and removed all the dependencies on log4j.

Here are the modifications to the pom.xml for setting dependencies.


<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>in.springframework.blog</groupId>
<artifactId>tutorials</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>tutorials</name>
<description>Demo project for Spring Boot</description>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.0.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
<swagger.version>2.9.2</swagger.version>
<gson.version>2.8.6</gson.version>
<log4j.version>2.13.3</log4j.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jetty</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-deploy</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-rewrite</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-security</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-expression</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.security</groupId>
<artifactId>spring-security-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.security</groupId>
<artifactId>spring-security-config</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-core -->
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>${log4j.version}</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>${gson.version}</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>${swagger.version}</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>${swagger.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.security</groupId>
<artifactId>spring-security-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>pl.project13.maven</groupId>
<artifactId>git-commit-id-plugin</artifactId>
<version>2.2.1</version>
</plugin>
</plugins>
</build>
</project>
view raw pom-0506.xml hosted with ❤ by GitHub
Now we look at our User entity. We need to add auditing fields to this table. Since in a realistic project one would have multiple entities, we create a abstract entity base class with all the audit fields. We also move the primary key to the abstract base class. When we do that, hibernate provides multiple ways of mapping the entities to the database schema. We want all the parent attributes to be stored in a single table with the child class attribute. To accomplish that, we need to add @MappedSuperClass to the base class.
package in.springframework.blog.tutorials.user.domain;
import lombok.Getter;
import lombok.Setter;
import org.springframework.data.annotation.CreatedBy;
import org.springframework.data.annotation.CreatedDate;
import org.springframework.data.annotation.LastModifiedBy;
import org.springframework.data.annotation.LastModifiedDate;
import javax.persistence.*;
import java.util.Date;
@Getter
@Setter
@MappedSuperclass
public abstract class AbstractBaseEntity {
@Id
@GeneratedValue(strategy= GenerationType.AUTO)
private Long id;
@CreatedBy
private Long createdBy;
@LastModifiedBy
private Long updatedBy;
@CreatedDate
private Date createdAt;
@LastModifiedDate
private Date updatedAt;
}
view raw AbstractBaseEntity.java hosted with ❤ by GitHub

As we can see, we have added five attributes in the base class. One is the id for all our entities and rest four will be used for auditing purposes.
At this time we also add another layer to our code. Currently all the Endpoints directly call the Repository layer, this causes a problem if we want to write functions that can be reused across different endpoints. An example of this need is retrieveUser method that takes an argument that could be a username or a email. Currently this method lies in the Endpoint layer as a private method. This is a useful method in many different contexts, so we create a new UserService layer and move this method there.

package in.springframework.blog.tutorials;
import in.springframework.blog.tutorials.user.domain.User;
import in.springframework.blog.tutorials.user.repository.UserRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.Optional;
@Service
public class UserService {
@Autowired
private UserRepository userRepository;
public Iterable<User> findAll() {
return userRepository.findAll();
}
public Optional<User> retrieveUser(String idOrUserNameOrEmail) {
try {
Long id = Long.parseLong(idOrUserNameOrEmail);
Optional<User> optionalUser = userRepository.findById(id);
if (optionalUser.isPresent()) {
return optionalUser;
}
}
catch(NumberFormatException e) {
}
Optional<User> optionalUser = userRepository.findUserByEmail(idOrUserNameOrEmail);
if (optionalUser.isPresent()) {
return optionalUser;
}
optionalUser = userRepository.findUserByUsername(idOrUserNameOrEmail);
return optionalUser;
}
public User save(User user) {
return userRepository.save(user);
}
public void delete(User user) {
userRepository.delete(user);
}
}
view raw UserService.java hosted with ❤ by GitHub

Now, let's get to the original task of enabling auditing. First we define a Auditing Config as below.
package in.springframework.blog.tutorials.configs;
import in.springframework.blog.tutorials.RequestContext;
import in.springframework.blog.tutorials.user.domain.User;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.domain.AuditorAware;
import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
import java.util.Optional;
@Configuration
@EnableJpaAuditing(auditorAwareRef = "auditorAware")
public class EntityAuditConfig {
@Bean
public AuditorAware<Long> auditorAware() {
return new AuditorAware<Long>() {
@Override
public Optional<Long> getCurrentAuditor() {
return Optional.of(RequestContext.currentUser.get().getId());
}
};
}
}
view raw EntityAuditConfig.java hosted with ❤ by GitHub

We had earlier defined a ThreadLocal that is used by the auditAware method defined above to extract currently logged in user and return its userId. As we can see the audit fields in the AbstractBaseEntity expects a Long for @createdBy and  @LastModifiedBy fields. The EntityAuditConfig also has annotation @EnableJpaAuditing which is required.
At this point we also add a new endpoint called ProfileEndpoint which can be used to manage the entity that represents a user profile. This entity currently only contains a url.
Now if we perform any operation on any of the endpoint, we will see the auditing fields automatically populated. Give it a spin. It is a life saver in many productions applications. I have had situations where users changed their passwords, forgot them and then complained saying that they have been hacked. 
The complete code for this and previous posts can be found at my github repository. This tutorial changes are under v1.5.
By at No comments:
Labels: , ,

Friday, April 12, 2019

Spring and more Kafka

In the last post, we saw how to integrate Kafka with Spring Boot application. The post was a very simple implementation of Kafka. The real world is much more complex. You have to deal with multiple topics, you need multiple partitions. In this post, we explore more details of a spring boot application with Kafka.
Let's start with the fact that we have multiple topics that we are dealing with. The very first thing that we need to do is to separate out the properties for each of the topics.
spring.kafka.topic1.bootstrap-servers=localhost:9092
spring.kafka.topic1.consumer.group-id=firstGroupId
spring.kafka.topic2.bootstrap-servers=localhost:9092
spring.kafka.topic2.consumer.group-id=secondGroupId
view raw application-kafka-multiple.properties hosted with ❤ by GitHub

We can no longer use default KafkaTemplate, we will have to create our own. The best way is to define a KafkaProducer. To create a KafkaProducer, we need to create KarkaProducerConfig. Here are two sets of KafkaProducerConfig class, one each for the topic.
package in.springframework.blog.tutorials.configs;
import in.springframework.blog.tutorials.Constants;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
@Component
public class FirstTopicProducerConfig {
@Value("${spring.kafka.topic1.bootstrap-servers:localhost}")
private String bootstrapAddress;
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
configProps.put(
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
configProps.put(
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean(name = Constants.FIRST_TOPIC_TEMPLATE_NAME)
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
view raw FirstTopicProducerConfig.java hosted with ❤ by GitHub

package in.springframework.blog.tutorials.configs;
import in.springframework.blog.tutorials.Constants;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
@Component
public class SecondTopicProducerConfig {
@Value("${spring.kafka.topic2.bootstrap-servers:localhost}")
private String bootstrapAddress;
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
configProps.put(
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
configProps.put(
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean(name = Constants.SECOND_TOPIC_TEMPLATE_NAME)
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
view raw SecondTopicProducerConfig.java hosted with ❤ by GitHub

Looking at the producer config, we can observe that we create two beans which are qualified by names and return an appropriate KafkaTemplate which is later used to send the message. Similar to producer config, we need to create consumer config. Consumer config is used by the listener to listen for the message.
package in.springframework.blog.tutorials.configs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import java.util.HashMap;
import java.util.Map;
@EnableKafka
@Configuration
public class FirstTopicConsumerConfig {
@Value("${spring.kafka.topic1.bootstrap-servers:localhost}")
private String bootstrapAddress;
@Value("${spring.kafka.topic1.group-id:firstgroup}")
private String groupId;
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
props.put(
ConsumerConfig.GROUP_ID_CONFIG,
groupId);
props.put(
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
props.put(
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean(name="firstTopicListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory
= new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
view raw FirstTopicConsumerConfig.java hosted with ❤ by GitHub

package in.springframework.blog.tutorials.configs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import java.util.HashMap;
import java.util.Map;
@EnableKafka
@Configuration
public class SecondTopicConsumerConfig {
@Value("${spring.kafka.topic2.bootstrap-servers:localhost}")
private String bootstrapAddress;
@Value("${spring.kafka.topic2.group-id:secondgroup}")
private String groupId;
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
props.put(
ConsumerConfig.GROUP_ID_CONFIG,
groupId);
props.put(
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
props.put(
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean(name="secondTopicListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory
= new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
view raw SecondTopicConsumerConfig.java hosted with ❤ by GitHub

As we can see in each of the consumer configs, we create a bean which returns a ConcurrentKafkaListenerContainerFactory. The beans are qualified by a name so that we can use an appropriate container factory for receiving messages.
We also modify the MyTopicMessage and add a member variable topicName that will help us distinguish the topic to which the message needs to be sent.
package in.springframework.blog.tutorials.pojos;
import in.springframework.blog.tutorials.Constants;
public class MyTopicMessage {
public MyTopicMessage() {
}
public MyTopicMessage(String message, Constants.TOPIC_NAME topicName) {
this.message = message;this.topicName = topicName;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
public Constants.TOPIC_NAME getTopicName() {
return topicName;
}
public void setTopicName(Constants.TOPIC_NAME topicName) {
this.topicName = topicName;
}
private String message;
private Constants.TOPIC_NAME topicName;
}
view raw MyTopicMessage.java hosted with ❤ by GitHub

We also modify the endpoint so that the message can be sent to the appropriate topic.
package in.springframework.blog.tutorials.endpoints;
import in.springframework.blog.tutorials.Constants;
import in.springframework.blog.tutorials.MyConstants;
import in.springframework.blog.tutorials.pojos.MyTopicMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.MediaType;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/send")
public class MyTopicMessageSenderEndpoint {
@Autowired
@Qualifier(Constants.FIRST_TOPIC_TEMPLATE_NAME)
private KafkaTemplate firstKafkaTemplate;
@Autowired
@Qualifier(Constants.SECOND_TOPIC_TEMPLATE_NAME)
private KafkaTemplate secondKafkaTemplate;
Logger logger = Logger.getLogger(MyTopicMessageSenderEndpoint.class);
@RequestMapping(method = RequestMethod.POST, produces = MediaType.APPLICATION_JSON_VALUE)
@PreAuthorize(MyConstants.ANNOTATION_ROLE_USER)
public String sendMessage(@RequestBody MyTopicMessage message) {
switch(message.getTopicName()) {
case FirstTopic:
firstKafkaTemplate.send(Constants.FIRST_TOPIC, message.getMessage()).addCallback(new ListenableFutureCallback() {
@Override
public void onFailure(Throwable throwable) {
logger.log(Level.INFO, String.format("Message '%s' failed!", message.getMessage()));
}
@Override
public void onSuccess(Object o) {
logger.log(Level.INFO, String.format("Message '%s' sent successfully!", message.getMessage()));
}
});
break;
case SecondTopic:
secondKafkaTemplate.send(Constants.SECOND_TOPIC, message.getMessage()).addCallback(new ListenableFutureCallback() {
@Override
public void onFailure(Throwable throwable) {
logger.log(Level.INFO, String.format("Message '%s' failed!", message.getMessage()));
}
@Override
public void onSuccess(Object o) {
logger.log(Level.INFO, String.format("Message '%s' sent successfully!", message.getMessage()));
}
});
break;
}
return "Success!";
}
}
view raw MyTopicMessageSenderEndpoint.java hosted with ❤ by GitHub

Now we modify the Listener to integrate everything so that the appropriate message can be received. Look at @KafkaListener annotation, we pass on the ListenerContainerFactory as an argument to receive a message from a queue.
package in.springframework.blog.tutorials.listeners;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
@Component
public class KafkaMessageListener {
Logger logger = LogManager.getLogger(KafkaMessageListener.class);
@KafkaListener(topics = "FirstTopic", containerFactory = "firstTopicListenerContainerFactory")
public void processFirstTopicMessage(@Payload String content,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
logger.info(String.format("Received %s message for partition %d ", topic, partition) + content);
}
@KafkaListener(topics = "SecondTopic", containerFactory = "secondTopicListenerContainerFactory")
public void processSecondTopicMessage(@Payload String content,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
logger.info(String.format("Received %s message for partition %d ", topic, partition) + content);
}
}
view raw KafkaMessageListener.java hosted with ❤ by GitHub

Now we can test the server. The flow of the service is as below.

  1. The message is posted to the endpoint as a POST request
  2. The @RestController receives the message and based on the topicName in the request, it sends to message to the topic with the same name.
  3. The listener receives the message from the appropriate queue.
$ curl -X POST \
>   'http://localhost:8081/send?token=3193fa24-a0ba-451b-83ff-eb563c3fd43b-cdf12811-7e41-474b-8fa6-e8fefd4a738c' \
>   -H 'Content-Type: application/json' \
>   -H 'Postman-Token: 15fbe075-9c80-4af9-a797-6b5e0979fd1b' \
>   -H 'cache-control: no-cache' \
>   -H 'token: 3193fa24-a0ba-451b-83ff-eb563c3fd43b-cdf12811-7e41-474b-8fa6-e8fefd4a738c' \
>   -d '{
> "message" : "This is my message!",
> "topicName" : "FirstTopic"
> }'
Success!

The message receipt is indicated in the server log.
2019-04-12 14:50:47.142  INFO 51396 --- [ntainer#0-0-C-1] i.s.b.t.listeners.KafkaMessageListener   : Received FirstTopic message for partition 0 This is my message!
Similarly, we can send a message to SecondTopic.
$ curl -X POST   'http://localhost:8?token=3193fa24-a0ba-451b-83ff-eb563c3fd43b-cdf12811-7e41-474b-8fa6-e8fefd4a738c'   -H 'Content-Type: application/json'   -H 'Postman-Token: 15fbe075-9c80-4af9-a797-6b5e0979fd1b'   -H 'cache-control: no-cache'   -H 'token: 3193fa24-a0ba-451b-83ff-eb563c3fd43b-cdf12811-7e41-474b-8fa6-e8fefd4a738c'   -d '{
"message" : "This is another message!",
"topicName" : "SecondTopic"
}'
Success!

The message receipt is indicated in the server log.
2019-04-12 14:53:14.972  INFO 51396 --- [ntainer#1-0-C-1] i.s.b.t.listeners.KafkaMessageListener   : Received SecondTopic message for partition 0 This is another message!
The complete code base for this tutorial can be found at my github repository at v1.4.
By at No comments:
Labels: , , , ,

Wednesday, April 10, 2019

Kafka and Spring

Kafka has become a very popular platform and is being used as a stream, journal and even eventing system. In this post, we explore how to integrate Kafka with spring framework application. First, we add the Kafka bootstrap server details in the application.properties file.
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=ofs
view raw application-kafka.properties hosted with ❤ by GitHub

Let's also add dependencies in pom.xml.
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
view raw pom-kafka.xml hosted with ❤ by GitHub

Now, for each Kafka topic, we create a listener class. The listener class provides a callback method that is called when any message is retrieved on that topic.
package in.springframework.blog.tutorials.listeners;
import in.springframework.blog.tutorials.Constants;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class MyTopicKafkaListener {
Logger logger = LogManager.getLogger(MyTopicKafkaListener.class);
@KafkaListener(topics = Constants.MY_TOPIC)
public void processMessage(String content) {
logger.info("Received Promise message " + content);
}
}
view raw MyTopicKafkaListener.java hosted with ❤ by GitHub

Now we create an endpoint through which we inject a message in the queue. The message is sent to the queue and is retrieved by the listener.

package in.springframework.blog.tutorials.endpoints;
import in.springframework.blog.tutorials.Constants;
import in.springframework.blog.tutorials.MyConstants;
import in.springframework.blog.tutorials.pojos.MyTopicMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/send")
public class MyTopicMessageSenderEndpoint {
@Autowired
private KafkaTemplate kafkaTemplate;
@RequestMapping(method = RequestMethod.POST, produces = MediaType.APPLICATION_JSON_VALUE)
@PreAuthorize(MyConstants.ANNOTATION_ROLE_USER)
public String createUser(@RequestBody MyTopicMessage message) {
kafkaTemplate.send(Constants.MY_TOPIC, message.getMessage());
return "Message sent successfully!.";
}
}
view raw MyTopicMessageSenderEndpoint.java hosted with ❤ by GitHub
We autowire a KafkaTemplate instance that is used to send the message to the queue.

$ curl -X POST \
>   'http://localhost:8081/send?token=3193fa24-a0ba-451b-83ff-eb563c3fd43b-cdf12811-7e41-474b-8fa6-e8fefd4a738c' \
>   -H 'Content-Type: application/json' \
>   -H 'Postman-Token: e281e3c5-0dae-4bb7-ac8d-6555f66a18c6' \
>   -H 'cache-control: no-cache' \
>   -H 'token: 3193fa24-a0ba-451b-83ff-eb563c3fd43b-cdf12811-7e41-474b-8fa6-e8fefd4a738c' \
>   -d '{
> "message" : "This is my message!"
> }'
Message sent successfully!.
The receipt of message is indicated in the spring server log.

2019-04-10 14:28:03.969  INFO 31091 --- [ntainer#0-0-C-1] i.s.b.t.listeners.MyTopicKafkaListener   : Received Promise message This is my message!
By at 1 comment:
Labels: , ,

Tuesday, February 12, 2019

12. Adding GIT release information

In the previous post, we saw how to enable actuator endpoints on our spring server. Once we have done that, it is a good idea to add GIT release information to the server in order to get the information related to the currently deployed server at runtime.
We add the git-commit-id-plugin to our pom.xml.

   <plugin>
    <groupid>pl.project13.maven</groupid>
    <artifactid>git-commit-id-plugin</artifactid>
    <version>2.2.1</version>
   </plugin>
The next things to do is to create a git.properties file in the resource directory of the project.
# git.properties
git.tags=${git.tags}
git.branch=${git.branch}
git.dirty=${git.dirty}
git.remote.origin.url=${git.remote.origin.url}
git.commit.id=${git.commit.id}
git.commit.id.abbrev=${git.commit.id.abbrev}
git.commit.id.describe=${git.commit.id.describe}
git.commit.id.describe-short=${git.commit.id.describe-short}
git.commit.user.name=${git.commit.user.name}
git.commit.user.email=${git.commit.user.email}
git.commit.message.full=${git.commit.message.full}
git.commit.message.short=${git.commit.message.short}
git.commit.time=${git.commit.time}
git.closest.tag.name=${git.closest.tag.name}
git.closest.tag.commit.count=${git.closest.tag.commit.count}
git.build.user.name=${git.build.user.name}
git.build.user.email=${git.build.user.email}
git.build.time=${git.build.time}
git.build.host=${git.build.host}
git.build.version=${git.build.version}
view raw git.properties hosted with ❤ by GitHub

The git-commit-id is added by the maven plugin so it is a good idea to build the project using maven.

$ mvn clean package -DskipTests
Now we can run the server and check the /manage/info endpoint using curl command.
$ curl -X GET http://localhost:9091/manage/info
{"git":{"branch":"master","commit":{"id":"6fb94c0","time":1549868235.000000000}}

By at No comments:
Labels: , , ,

11. Spring Actuators

Spring provides actuators that are a helpful set of tools to debug the application on runtime. Here is how to enable them. We first add actuator dependency in the pom.xml
<dependency>
  <groupid>org.springframework.boot</groupid>
  <artifactid>spring-boot-starter-actuator</artifactid>
</dependency>

Now we define a prefix for all the actuator endpoints. We add the following line into the application.properties file. This enables all the actuator endpoints. We can enable specific endpoints by adding a comma delimited list of endpoints. We also deploy management endpoint on a separate port so that we can block its access from something like ELB.
management.endpoints.web.base-path=/manage
management.server.port=9091
management.endpoints.web.exposure.include=*
Since we already have a security filter defined, we need to exempt health and info endpoint from security check. We add the following URLs int he SecurityConfiguration configure method.
@Override
    public void configure(WebSecurity web) throws Exception {
        web.ignoring().antMatchers("/manage/health");
        web.ignoring().antMatchers("/manage/info");
        web.ignoring().antMatchers("/webjars/**");
        web.ignoring().antMatchers("/error");
        web.ignoring().antMatchers("/swagger-ui.html");
        web.ignoring().antMatchers("/v2/api-docs/**");
        web.ignoring().antMatchers("/swagger-resources/**");
    }
Here we have added paths related to error, actuator, and swagger.
This enables actuator endpoints for our server. We can query these endpoints and following is the sample response.

$ curl -X GET http://localhost:9091/manage/health
{"status":"UP"}
By at No comments:
Labels: ,
Subscribe to: Posts (Atom)