DEV Community: eidher

Kafka Connect: FileStreamSourceConnector in distributed mode

eidher — Fri, 23 Aug 2024 22:31:57 +0000

1) Source connectors: Start our kafka cluster. This is the docker-compose.yml file:

version: '2'

services:

  kafka-cluster:
    image: landoop/fast-data-dev:cp3.3.0
    environment:
      ADV_HOST: 127.0.0.1
      RUNTESTS: 0
    ports:
      - 2181:2181                 # Zookeeper
      - 3030:3030                 # Landoop UI
      - 8081-8083:8081-8083       # REST Proxy, Schema Registry, Kafka Connect
      - 9581-9585:9581-9585       # JMX Ports
      - 9092:9092                 # Kafka Broker

Create and start the container:

docker-compose up kafka-cluster

2) Create the topic we're going to write to:

docker run --rm -it --net=host landoop/fast-data-dev:cp3.3.0 bash
kafka-topics --create --topic demo-2-distributed --partitions 3 --replication-factor 1 --zookeeper 127.0.0.1:2181

3) In a browser go to 127.0.0.1:3030 -> Connect UI
Create a new connector -> File Source
Paste the next configuration and click the Create button:

name=file-stream-demo-distributed
connector.class=org.apache.kafka.connect.file.FileStreamSourceConnector
tasks.max=1
file=demo-file.txt
topic=demo-distributed
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true

4) Create the source file:
Execute docker ps to see the containerId.

docker exec -it <containerId> bash
touch demo-file.txt
echo "add any content" >> demo-file.txt

3) In a browser go to 127.0.0.1:3030 -> Kafka Topics UI
Click the topic defined previously (demo-distributed) and validate the content:

Distributed Tracing

eidher — Tue, 21 Feb 2023 17:35:12 +0000

Spring Cloud Sleuth provides Spring Boot auto-configuration for distributed tracing.
https://spring.io/projects/spring-cloud-sleuth#overview

Zipkin is a distributed tracing system.
https://zipkin.io

Dependencies

<dependency>
  <artifactId>spring-cloud-starter-sleuth</artifactId>
  <groupId>org.springframework.cloud</groupId>
</dependency>

<dependency>
  <artifactId>spring-cloud-sleuth-zipkin</artifactId>
  <groupId>org.springframework.cloud</groupId>
</dependency>

<dependency>
  <artifactId>spring-rabbit</artifactId>
  <groupId>org.springframework.amqp</groupId>
</dependency>

Properties

spring.sleuth.sampler.probability=1.0

Now, if you have zipkin and rabbitmq running, you can call your service and see the traces in zipkin.

For more information see https://www.appsdeveloperblog.com/spring-cloud-sleuth-and-zipkin/

Resilience4J Circuit Breakers

eidher — Mon, 20 Feb 2023 16:22:00 +0000

If we have a chain of services calling other services and one of the services is down or slow, it would impact the entire chain. In that case, we would need to return a fallback response. We could implement a circuit breaker pattern to reduce the load, retry the requests, and implement a rate limiting.

Resilience4j is a lightweight fault tolerance library designed for functional programming.
https://resilience4j.readme.io/docs

Dependencies

<dependency>
  <artifactId>resilience4j-spring-boot2</artifactId>
  <groupId>io.github.resilience4j</groupId>
</dependency>
<dependency>
  <artifactId>spring-boot-starter-aop</artifactId>
  <groupId>org.springframework.boot</groupId>
</dependency>
<dependency>
  <artifactId>spring-boot-starter-actuator</artifactId>
  <groupId>org.springframework.boot</groupId>
</dependency>

Retry

When an API call fails, you can force a retry just by adding the @Retry annotation to the API endpoint. By default, it would retry three times, but you can set the max retry attempts by giving the retry a name and adding a max attempts property:

@GetMapping("/sample-api")
@Retry(name = "sample-api")
public String sampleApi() {
  ...
}

resilience4j.retry.instances.sample-api.max-attempts=5

Additionally, you can define other properties as wait duration (between attempts), and enable exponential backoff to wait in exponential ranges of time:

resilience4j.retry.instances.sample-api.waitDuration=1s
resilience4j.retry.instances.sample-api.enableExponentialBackoff=true

Fallback

You can define a fallback method in case after retries it fails again:

@GetMapping("/sample-api")
@Retry(name = "sample-api", fallbackMethod = "hardcodedResponse")
public String sampleApi() {
  ...
}

public String hardcodedResponse(Exception ex) {
  return "fallback-response";
}

Circuit breaker

@GetMapping("/sample-api")
@CircuitBreaker(name = "default", fallbackMethod = "hardcodedResponse")
public String sampleApi() {
  ...
}

public String hardcodedResponse(Exception ex) {
  return "fallback-response";
}

With this configuration, if sampleApi method fails, the fallback method is executed, for the next calls it will break the circuit and it will directly return the fallback response.

The CircuitBreaker is implemented via a finite state machine with three normal states: CLOSED, OPEN, and HALF_OPEN.
https://resilience4j.readme.io/docs/circuitbreaker

Closed is when I am calling the microservice continuously. In the open state, the circuit breaker will not call the dependent microservice. It will return the fallback response. Finally, in the HALF_OPEN state, the circuit breaker will be sending a percentage of requests to the dependent microservice, and for the rest of the requests, it will return the fallback response. When starting the application the circuit breaker starts in the closed state. If a significant percentage of the calls are failing, the circuit breaker will switch to the open state. There is a wait duration that you can configure, then it will switch to HALF_OPEN state. You can configure how much percentage of requests it will try to send successfully to the dependant service to move to the closed state again. If it fails, it will move back to the open state again.

Example of a configuration for failure rate threshold in percentage, the default value is 50%:

resilience4j.circuitbreaker.instances.default.failureRateThreshold=90

For more information see: https://www.baeldung.com/spring-boot-resilience4j

Rate Limiting

The @RateLimiter annotation allows us to define a period for a specific number of calls. If you exceed that quantity of calls, the service will return an exception.

@GetMapping("/sample-api")
@RateLimiter(name="default")
public String sampleApi() {
  ...
}

resilience4j.ratelimiter.instances.default.limitForPeriod=2
resilience4j.ratelimiter.instances.default.limitRefreshPeriod=10s

BulkHead

The @Bulkhead annotation allows us to define a max concurrent calls value.

@GetMapping("/sample-api")
@Bulkhead(name = "sample-api")
public String sampleApi() {
  ...
}

resilience4j.bulkhead.instances.sample-api.maxConcurrentCalls=10

Spring Cloud Gateway

eidher — Sun, 19 Feb 2023 16:32:29 +0000

Spring Cloud Gateway aims to provide a simple, yet effective way to route to APIs and provide cross-cutting concerns to them such as security, monitoring/metrics, and resiliency.
https://spring.io/projects/spring-cloud-gateway

First, configure the project as a Eureka client as was explained here. Then, add this dependency:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-gateway</artifactId>
</dependency>

If you add the next properties you can access other services registered in Eureka:

spring.cloud.gateway.discovery.locator.enabled=true
spring.cloud.gateway.discovery.locator.lowerCaseServiceId=true

The second property is optional and allows us to use lowercase to access the service. For instance, if you have a eureka client service named ACCOUNTS-MICROSERVICE, and the API Gateway project is running on port 8765 you can use this url: http://localhost:8765/accounts-microservice/accounts-microservice/accountDetails/1. Now, you are accessing the accounts-microservice service through the API gateway.

You can create a custom router for adding headers, parameters, and routing:

@Bean
public RouteLocator gatewayRouter(RouteLocatorBuilder builder) {
  return builder.routes()
      .route(p -> p.path("/accounts-microservice/**")
          .filters(f -> f
              .addRequestHeader("MyHeader", "MyURI")
              .addRequestParameter("Param", "MyValue"))
          .uri("lb://accounts-microservice"))
      .build();
}

Now, you can remove the two previous properties and use this simpler URL: http://localhost:8765/accounts-microservice/accountDetails/1. You can see the new header and parameter in the request. Besides, you can chain as many routes as you want in the same builder, add path rewrites, etc.

Additionally, Spring Cloud Gateway provides a GlobalFilter to handle cross-cutting concerns such as logging, lets see an example:

@Component
public class LoggingFilter implements GlobalFilter {

  private final Logger logger = LoggerFactory.getLogger(LoggingFilter.class);

  @Override
  public Mono<Void> filter(ServerWebExchange exchange,
      GatewayFilterChain chain) {
    logger.info("Path of the request received -> {}",
        exchange.getRequest().getPath());
    return chain.filter(exchange);
  }

}

For more information see: https://www.baeldung.com/spring-cloud-gateway

Spring Cloud OpenFeign

eidher — Thu, 16 Feb 2023 21:02:25 +0000

Feign is a declarative web service client. It makes writing web service clients easier. To use Feign create an interface and annotate it.

Source: https://cloud.spring.io/spring-cloud-netflix/multi/multi_spring-cloud-feign.html

With Feign, URLs are not hardcoded and Feign integrates with Ribbon and Eureka automatically. To use it you must add this dependency:

<dependency>
  <artifactId>spring-cloud-starter-openfeign</artifactId>
  <groupId>org.springframework.cloud</groupId>
</dependency>

Then, you must add the @EnableFeignClients annotation to the main class and declare a Feign client using the @FeignClient annotation in the interface:

@FeignClient(name = "other-service-name")
public interface Proxy {

  @GetMapping("/other-service-uri/{variable}")
  public Response getResponse(
      @PathVariable String variable);

}

Now, you can use it:

Response response = proxy.getResponse(variable);

For more information see: https://www.baeldung.com/spring-cloud-openfeign

Spring Cloud Config

eidher — Wed, 15 Feb 2023 23:59:50 +0000

Spring Cloud Config provides server-side and client-side support for externalized configuration in a distributed system. With the Config Server, you have a central place to manage external properties for applications across all environments.

Source: https://docs.spring.io/spring-cloud-config/docs/current/reference/html/

Spring Cloud Config Server

Spring Cloud Config Server provides an HTTP resource-based API for external configuration (name-value pairs or equivalent YAML content). The server is embeddable in a Spring Boot application, by using the @EnableConfigServer annotation. Consequently, the following application is a config server:

@SpringBootApplication
@EnableConfigServer
public class ConfigServer {
  public static void main(String[] args) {
    SpringApplication.run(ConfigServer.class, args);
  }
}

Source: https://docs.spring.io/spring-cloud-config/docs/current/reference/html/#_spring_cloud_config_server

The config server application needs this dependency:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-config-server</artifactId>
</dependency>

The default implementation of the server storage backend uses git, but you can also use SVN, file system, CredHub, AWS (Secrets Manager, Parameter Store, or S3), JDBC, Redis, or Vault. For git, you must create a repository containing YAML and properties files in the master branch. Then, define it in the application.properties file of the Config Server application:

spring.application.name=spring-cloud-config-server
server.port=8888
spring.cloud.config.server.git.uri=file:///C:/Users/home/Desktop/yourProject/git-repo

If you added a config-client.properties file, for instance, in the git repository, you can run the config server application and access http://localhost:8888/config-client/default to view the properties (default is the profile).

Spring Cloud Config Client

Spring Boot applications can use the Spring Config Server. The config client applications need this dependency:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-config</artifactId>
</dependency>

Then, add in the application.properties file of the Config client applications this property:

spring.config.import=optional:configserver:http://localhost:8888

Additionaly, if you are using different profiles you must define the profile and it will use the corresponding properties file. Add this property: spring.cloud.config.profile=profile (where profile is dev, qa, etc)

Finally, you can use the @ConfigurationProperties or @Value annotations to access the properties from the config server.

For more information see: https://www.baeldung.com/spring-cloud-configuration

Handling Injection Attacks in Java

eidher — Sun, 14 Mar 2021 03:21:53 +0000

An injection attack is the insertion of malicious data from the client to the application using SQL or XXE (XML External Entity).
It is important to prevent injection attacks because it allows attackers to spoof identity, tamper with existing data, disclosure all the data, destroy the data, become the administrator, etc.

SQL Injection

If the attacker introduces something like ' or 1=1 -- the application could display data from the database:

A patch to the SQL injection

The flaw is because the field (accountName) is concatenated to the SQL statement:

String query = "SELECT * FROM user_data WHERE last_name = '" + accountName + "'";

Statement statement = connection.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,
                        ResultSet.CONCUR_READ_ONLY);
ResultSet results = statement.executeQuery(query);

To patch this flaw you must use Immutable Queries, Static Queries, Parameterized Queries, or Stored Procedures. For the previous example the best solution is a parameterized query:

final String query = "SELECT * FROM user_data WHERE last_name = ?";

try {
  PreparedStatement statement = connection.prepareStatement(query,
                    ResultSet.TYPE_SCROLL_INSENSITIVE,
                    ResultSet.CONCUR_READ_ONLY);
  statement.setString(1, accountName);
  ResultSet results = statement.executeQuery(query);
  ...
} catch (SQLException sqle) {
  ...
}

Now, if you try to inject SQL you get an exception.

XXE Injection

If you have a service that receives an XML, somebody could change that XML (using Burp Suite) in order to access local resources, execute code remotely, disclose files, or execute a DoS attack (using a Billion laughs attack).

You may say "no worries, I use JSON in my REST services". However, the attacker could change the content type of the request body and send the same XML. For instance, next we have the request of a service caught by Burp Suite (ellipsis is used to omit irrelevant information):

...
Content-Type: application/json
...

{"text":"test"}

It is as easy as to change the Content-Type to XML and the payload to perform the attack:

...
Content-Type: application/xml
...

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE foo [
  <!ELEMENT foo ANY>
  <!ENTITY xxe SYSTEM "../../">
]>
<comment>
  <text>&xxe;</text>
</comment>

This service is used to post a comment, but now we have posted a directory of the server:

After some tries, you could print the content of an important file (like passwords or configurations).

A patch to the XXE injection

You can validate the input, the content type, or instruct your parser to ignore DTD (document type definition). See the setProperty invocation:

protected Comment parseXml(String xml) throws Exception {
  JAXBContext jc = JAXBContext.newInstance(Comment.class);

  XMLInputFactory xif = XMLInputFactory.newFactory();
  xif.setProperty(XMLInputFactory.SUPPORT_DTD, false);

  XMLStreamReader xsr = xif.createXMLStreamReader(new StringReader(XML));

  Unmarshaller unmarshaller = jc.createUnmarshaller();
  return (Comment) unmarshaller.unmarshal(xsr);
}

For Spring REST Services, you can specify the consumes = MediaType.APPLICATION_JSON_VALUE:

@RequestMapping(method = RequestMethod.POST, consumes = MediaType.APPLICATION_JSON_VALUE, produces = MediaType.APPLICATION_JSON_VALUE)
@ResponseBody
public AttackResult createNewUser(@RequestBody String commentStr, @RequestHeader("Content-Type") String contentType) throws Exception {
        ...
}

Now, the attacker cannot send XML:

For more information see: https://cheatsheetseries.owasp.org/cheatsheets/XML_External_Entity_Prevention_Cheat_Sheet.html

Handling Cross-Site Scripting (XSS) in Java

eidher — Fri, 26 Feb 2021 04:15:46 +0000

Cross-site scripting (XSS) is a type of security vulnerability in web applications where an attacker injects malicious scripts through some kind of user input (like input boxes, URL parameters, HTML headers, etc)
It is important to prevent XSS attacks to safeguard the confidentiality, integrity, and availability of the information of the web application. The two main cross-site scripting flaws are reflected and stored:

Reflected XSS

Malicious content from a user request is displayed to the user or it is written into the page after from server response. For instance, in the next screenshot, the credit card number field is vulnerable. After the number, there is a script to be injected:



<script>alert('my javascript here')</script>

When the purchase button is clicked, the alert windows is displayed:

A patch to the flaw in Java

When you have a String RequestParam, avoid handling it without sanitization:

The OWASP Java encoder has a method called forHtml for that purpose:

Now, the field is printed as text, but it is not executed:

Stored XSS

The payload is persisted. For example, in the next screenshot, you can see that a script is added as a comment. When the page is loaded the script is executed and printed as part of the code.

A patch to the flaw in Java

The solution is to sanitize the RequestBody before handling it:

Now, the comment is printed as text, but it is not executed:

Testing with Mockito

eidher — Thu, 14 Jan 2021 05:22:57 +0000

1) Register MockitoExtension

@ExtendWith(MockitoExtension.class)
class ObjectTest {
    static final Long ID = 1L;

2) Create the mock

    @Mock
    private ObjectRepo mockRepo;

3) Inject the mock

    @InjectMocks
    private ObjectService objectService;

    @Test
    void whenfindByIdThenReturnResult() {
        var objectDAO = new ObjectDAO();
        objectDAO.setId(ID);

4) Define the behavior

        when(mockRepo.findById(any(Long.class))).thenReturn(Optional.of(objectDAO));

5) Test

        var result = ObjectService.findById(ID);

6) Verify

        verify(mockRepo, times(1)).findById(any(Long.class));

7) Validate

        assertAll(
                () -> assertNotNull(result),
                () -> assertEquals(objectDAO.getId(), result.getId())
        );
    }
}

Spring Injection Types

eidher — Mon, 11 Jan 2021 06:30:56 +0000

Spring supports three types of dependency injections:

Constructor injection

@Component
public class SecondBeanImpl implements SecondBean {

    private FirstBean firstBean;

    @Autowired
    public SecondBeanImpl(FirstBean firstBean) {
        this.firstBean = firstBean;
    }
}

That is similar to:

FirstBean firstBean = new FirstBeanImpl();
SecondBean secondBean = new SecondBeanImpl(firstBean);

This type of dependency injection instantiates and initializes the object.
In this approach, beans are immutable and dependencies are not null. However, if you define many parameters in the constructor, your code is not clean.
From Spring 4.3 the @Autowired annotation is not required if the class has a single constructor.

Setter injection

@Component
public class SecondBeanImpl implements SecondBean {

    private FirstBean firstBean;

    @Autowired
    public setFirstBean(FirstBean firstBean) {
        this.firstBean = firstBean;
    }
}

That is similar to:

FirstBean firstBean = new FirstBeanImpl();
SecondBean secondBean = new SecondBeanImpl();
secondBean.setFirstBean(firstBean);

In this approach, beans are not immutable (the setter could be called later), and not mandatory dependencies can lead to NullPointerExceptions.

Field injection

@Component
public class SecondBeanImpl implements SecondBean {

    @Autowired
    private FirstBean firstBean;
}

This approach may look cleaner but hides the dependencies and makes testing difficult. While constructor and setter injections use proxies, field injection uses reflection which could affect the performance. Could be used in test classes.

Jenkins: Automating your delivery pipeline

eidher — Mon, 28 Dec 2020 22:11:23 +0000

In this post, we are going to create a Jenkins pipeline for a Java Spring project using Maven and Git:

Accessing Jenkins dashboard and configuring Maven in Jenkins:

Click on "Manage Jenkins" and then in "Global Tool Configuration"

Add a Maven installation and provide a name for it. Click on Save.

Creation of our first build job in Jenkins for a Spring application:

Click on "New Item"

Provide a name for the project and choose "Freestyle Project"

Go to "Source Code Management", choose "Git", and provide the Repository URL and the branch.

Go to the "Build" tab and choose "Invoke top-level Maven targets"

Choose the Maven Version, write compile in the goals and click on "Save"

Click on "Build Now"

Click on "Console Output" in the first build of the Build History section

Confirm the "SUCCESS" message is displayed when finished

Developing and viewing of a delivery pipeline of a Spring application with two stages build and test in Jenkins

Let's create a New Item

Enter an item name, choose the "Freestyle project" and click OK

Go to the "Source Code Management" tab, select "Git", provide the URL of the Repository, and the branch

Go to the "Build" tab and choose "Invoke top-level Maven targets"

Choose the Maven Version, write test in the goals and click on "Save"

Click on "Build Now"

Click on "Console Output" in the first build of the Build History section

Confirm the "SUCCESS" message is displayed when finished

To do it automatically after the compile job, let's go to the "Configure" option of the item

Go to the "Build Triggers" section and select the "Build after other projects are built" option. Then, enter the other project name, and click "Save"

Go to the "Manage Jenkins" option in the main menu and choose "Manage Plugins"

Go to the "Available" tab and look for "Build pipeline". Choose it and click on "Install without Restart"

Choose the plus symbol in the main dashboard

Choose the "Build Pipeline View" option, provide a view name, and click on "OK"

Select the initial job and click ok

Click "Run"

Transforming your delivery pipeline in the form of scripted/declarative pipeline

First, install the "Pipeline Maven Integration Plugin"

Now, let's create a new item (Pipeline)

Then, go to the "Pipeline" tab and choose "Scripted Pipeline" or "GitHub + Maven" for a declarative pipeline

Finally, adapt the code according to your needs, and click on "Save"

If you click on "Build Now" you can see the result

Developing a Jenkinsfile for our application

If you chose "GitHub + Maven" for a declarative pipeline you can copy that code into a Jenkinsfile in the root of your project. Now, edit the Pipeline and select "Pipeline script from SCM". Then, choose Git as your SCM and provide the Repository URL and branch. Finally, click "Save" and "Build now" to test your pipeline

Spring Security

eidher — Sat, 21 Nov 2020 22:02:05 +0000

There are many authentication mechanisms (basic, digest, form, X.509, etc), and there are many storage options for credentials and authority information (in-memory, database, LDAP, etc). Authorization depends on authentication and determines if you have the required Authority. The decision process is often based on roles (e.g. ADMIN, MEMBER, GUEST, etc).

There are three steps to set up and configure Spring Security in a web environment:

Setup the filter chain: The implementation is a chain of Spring configured filters (Spring Boot does it automatically)
Configure security (authorization) rules
Setup Web Authentication

In the next example, it is defined as specific authorization restrictions for URLs using mvcMatchers.

@Configuration
@EnableWebSecurity
public class SecurityConfig extends WebSecurityConfigurerAdapter {

    @Override
    protected void configure(HttpSecurity http) throws Exception {
        http.formLogin().loginPage("/login").permitAll().and().exceptionHandling().accessDeniedPage("/denied").and()
                .authorizeRequests().mvcMatchers("/accounts/resources/**").permitAll().mvcMatchers("/accounts/edit*")
                .hasRole("EDITOR").mvcMatchers("/accounts/account*").hasAnyRole("VIEWER", "EDITOR")
                .mvcMatchers("/accounts/**").authenticated().and().logout().permitAll().logoutSuccessUrl("/");
    }

    @Autowired
    public void configureGlobal(AuthenticationManagerBuilder auth) throws Exception {
        auth.inMemoryAuthentication().passwordEncoder(new StandardPasswordEncoder()).withUser("viewerUser")
                .password("abc").roles("VIEWER").and().withUser("editorUser").password("abc").roles("EDITOR");
    }

}

As you can see, you can chain multiple restrictions (using the and() method). The first method sets up a form-based authentication. The second method uses a UserDetailsManagerConfigurer. You can use jdbcAuthentication() instead of inMemoryAuthentication().

Learn more at Spring Security Reference and Security Architecture with Spring