feat : enable and disable kafka listener #600
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,61 @@ | ||||||
| package com.example.springbootkafkasample.config; | ||||||
|
|
||||||
| import java.util.Comparator; | ||||||
| import java.util.List; | ||||||
| import java.util.Objects; | ||||||
| import org.springframework.core.Ordered; | ||||||
| import org.springframework.core.annotation.Order; | ||||||
| import org.springframework.http.HttpStatus; | ||||||
| import org.springframework.http.HttpStatusCode; | ||||||
| import org.springframework.http.ProblemDetail; | ||||||
| import org.springframework.http.converter.HttpMessageNotReadableException; | ||||||
| import org.springframework.validation.FieldError; | ||||||
| import org.springframework.web.bind.MethodArgumentNotValidException; | ||||||
| import org.springframework.web.bind.annotation.ControllerAdvice; | ||||||
| import org.springframework.web.bind.annotation.ExceptionHandler; | ||||||
| import org.springframework.web.bind.annotation.ResponseStatus; | ||||||
|
|
||||||
| @Order(Ordered.HIGHEST_PRECEDENCE) | ||||||
| @ControllerAdvice | ||||||
| class GlobalExceptionHandler { | ||||||
|
|
||||||
| @ExceptionHandler(MethodArgumentNotValidException.class) | ||||||
| @ResponseStatus(HttpStatus.BAD_REQUEST) | ||||||
| ProblemDetail onException(MethodArgumentNotValidException methodArgumentNotValidException) { | ||||||
|
There was a problem hiding this comment. 🛠️ Refactor suggestion Declare the The Apply this diff to change the method's access modifier: -ProblemDetail onException(MethodArgumentNotValidException methodArgumentNotValidException) {
+public ProblemDetail onException(MethodArgumentNotValidException methodArgumentNotValidException) {📝 Committable suggestion
Suggested change
|
||||||
| ProblemDetail problemDetail = | ||||||
| ProblemDetail.forStatusAndDetail(HttpStatusCode.valueOf(400), "Invalid request content."); | ||||||
| problemDetail.setTitle("Constraint Violation"); | ||||||
| List<ApiValidationError> validationErrorsList = methodArgumentNotValidException.getAllErrors().stream() | ||||||
| .map(objectError -> { | ||||||
| FieldError fieldError = (FieldError) objectError; | ||||||
| return new ApiValidationError( | ||||||
| fieldError.getObjectName(), | ||||||
| fieldError.getField(), | ||||||
| fieldError.getRejectedValue(), | ||||||
| Objects.requireNonNull(fieldError.getDefaultMessage(), "")); | ||||||
| }) | ||||||
| .sorted(Comparator.comparing(ApiValidationError::field)) | ||||||
| .toList(); | ||||||
| problemDetail.setProperty("violations", validationErrorsList); | ||||||
| return problemDetail; | ||||||
| } | ||||||
|
|
||||||
| @ExceptionHandler(IllegalArgumentException.class) | ||||||
| ProblemDetail handleIllegalArgumentException(IllegalArgumentException ex) { | ||||||
|
There was a problem hiding this comment. 🛠️ Refactor suggestion Declare the Similar to the previous method, Apply this diff to change the method's access modifier: -ProblemDetail handleIllegalArgumentException(IllegalArgumentException ex) {
+public ProblemDetail handleIllegalArgumentException(IllegalArgumentException ex) {📝 Committable suggestion
Suggested change
|
||||||
| ProblemDetail problemDetail = ProblemDetail.forStatusAndDetail(HttpStatusCode.valueOf(400), ex.getMessage()); | ||||||
| problemDetail.setTitle("Bad Request"); | ||||||
| return problemDetail; | ||||||
| } | ||||||
|
|
||||||
| @ExceptionHandler(HttpMessageNotReadableException.class) | ||||||
| public ProblemDetail handleInvalidEnum(HttpMessageNotReadableException ex) { | ||||||
| if (ex.getMessage() | ||||||
| .contains("Cannot deserialize value of type `com.example.springbootkafkasample.dto.Operation`")) { | ||||||
| return ProblemDetail.forStatusAndDetail( | ||||||
| HttpStatusCode.valueOf(400), "Invalid operation value. Allowed values are: START, STOP."); | ||||||
| } | ||||||
| return ProblemDetail.forStatusAndDetail(HttpStatusCode.valueOf(400), "Invalid request."); | ||||||
| } | ||||||
|
|
||||||
| record ApiValidationError(String object, String field, Object rejectedValue, String message) {} | ||||||
| } | ||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -8,6 +8,7 @@ | |
| import org.springframework.kafka.annotation.KafkaListenerConfigurer; | ||
| import org.springframework.kafka.config.KafkaListenerEndpointRegistrar; | ||
| import org.springframework.kafka.listener.KafkaListenerErrorHandler; | ||
| import org.springframework.lang.NonNull; | ||
|
There was a problem hiding this comment. 💡 Codebase verification Addition of @nonnull is safe but needs to be applied consistently The verification revealed another implementation of
🔗 Analysis chainLGTM! Good addition of null-safety. The addition of Let's verify that this change doesn't conflict with any custom implementations: Also applies to: 27-27 🏁 Scripts executedThe following scripts were executed for the analysis: Script: #!/bin/bash
# Description: Check for any custom implementations of KafkaListenerConfigurer
# that might be affected by the stricter null check
# Search for KafkaListenerConfigurer implementations
ast-grep --pattern 'class $_ implements $$$KafkaListenerConfigurer$$$'
# Search for direct calls to configureKafkaListeners
rg -l "configureKafkaListeners.*registrar"
Length of output: 349 Script: #!/bin/bash
# Let's check the implementation in the other KafkaConfig file to see if it also has @NonNull
# and verify there are no conflicts
# Check the implementation in boot-multiple-producers-consumers
cat kafka-spring-boot/boot-multiple-producers-consumers/src/main/java/com/example/springbootkafka/multi/config/KafkaConfig.java
# Also check if there are any other files importing KafkaListenerConfigurer
# to ensure we haven't missed any implementations
rg -l "import.*KafkaListenerConfigurer"
Length of output: 5852 |
||
| import org.springframework.validation.beanvalidation.LocalValidatorFactoryBean; | ||
|
|
||
| @Configuration | ||
|
|
@@ -23,7 +24,7 @@ public KafkaConfig(LocalValidatorFactoryBean validator) { | |
| } | ||
|
|
||
| @Override | ||
| public void configureKafkaListeners(@NonNull KafkaListenerEndpointRegistrar registrar) { | ||
| registrar.setValidator(this.validator); | ||
| } | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| package com.example.springbootkafkasample.dto; | ||
|
|
||
| import jakarta.validation.constraints.NotBlank; | ||
| import jakarta.validation.constraints.NotNull; | ||
|
|
||
| public record KafkaListenerRequest( | ||
| @NotBlank(message = "Container ID must not be blank") String containerId, | ||
| @NotNull(message = "Operation must not be null") Operation operation) {} | ||
|
Comment on lines
+1
to
+8
There was a problem hiding this comment. 💡 Codebase verification Security controls needed for Kafka listener management endpoint Based on the verification results:
Required security improvements:
🔗 Analysis chainVerify security implications and Operation enum usage The ability to start/stop Kafka listeners could have security implications. Consider:
Let's verify the Operation enum implementation and usage: 🏁 Scripts executedThe following scripts were executed for the analysis: Script: #!/bin/bash
# Find and display the Operation enum definition
echo "Looking for Operation enum definition:"
ast-grep --pattern 'enum Operation {
$$$
}'
# Find all usages of KafkaListenerRequest to verify security measures
echo -e "\nChecking for security controls in endpoints using KafkaListenerRequest:"
rg -l "KafkaListenerRequest" | xargs rg -A 5 "@(PreAuthorize|Secured|RolesAllowed)"
Length of output: 377 Script: #!/bin/bash
# Let's try a different approach to find the Operation enum
echo "Looking for Operation enum definition:"
fd -e java | xargs rg -l "enum Operation"
echo -e "\nChecking Operation enum content:"
fd -e java | xargs rg -A 5 "enum Operation"
echo -e "\nChecking KafkaListenerRequest usage in controllers/services:"
fd -e java | xargs rg -B 2 -A 5 "KafkaListenerRequest"
echo -e "\nChecking for any security configurations:"
fd -e java -e xml -e yml -e yaml -e properties | xargs rg "security|Security|@Secured|@PreAuthorize|WebSecurity"
Length of output: 13815 There was a problem hiding this comment. @coderabbitai, please help me to implement above security improvements There was a problem hiding this comment. @rajadilipkolli, I'd be happy to help you implement the security improvements. Here are the steps to enhance the security of your 1. Add Spring Security with Role-Based Access Control a. Add Spring Security Dependency Include the Spring Security starter dependency in your <!-- Add Spring Security starter -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-security</artifactId>
</dependency>b. Create a Security Configuration Class Create a new class package com.example.springbootkafkasample.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.method.configuration.EnableMethodSecurity;
import org.springframework.security.config.Customizer;
import org.springframework.security.config.annotation.web.configuration.WebSecurityCustomizer;
import org.springframework.security.web.SecurityFilterChain;
import static org.springframework.security.config.Customizer.withDefaults;
@Configuration
@EnableMethodSecurity
public class SecurityConfig {
@Bean
public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {
http
.authorizeHttpRequests(authorize -> authorize
.requestMatchers("/listeners/**").hasRole("ADMIN")
.anyRequest().permitAll()
)
.httpBasic(withDefaults())
.csrf().disable(); // Disable CSRF for simplicity; consider enabling it in production.
return http.build();
}
@Bean
public WebSecurityCustomizer webSecurityCustomizer() {
return (web) -> web.debug(false);
}
}c. Configure an In-Memory User for Testing For demonstration purposes, you can set up an in-memory user with the import org.springframework.context.annotation.Bean;
import org.springframework.security.provisioning.InMemoryUserDetailsManager;
import org.springframework.security.core.userdetails.User;
@Bean
public InMemoryUserDetailsManager userDetailsService() {
return new InMemoryUserDetailsManager(
User.withUsername("admin")
.password("{noop}adminpass") // {noop} specifies that no encoding is required
.roles("ADMIN")
.build()
);
}d. Secure the Controller Methods Ensure that only users with the package com.example.springbootkafkasample.controller;
import org.springframework.security.access.prepost.PreAuthorize;
//...
@RestController
@RequestMapping("/listeners")
public class MessageRestController {
//...
@PreAuthorize("hasRole('ADMIN')")
@GetMapping
public ResponseEntity<Map<String, Boolean>> getListeners() {
// Existing code
}
@PreAuthorize("hasRole('ADMIN')")
@PostMapping
public ResponseEntity<Map<String, Boolean>> updateListenerState(@RequestBody KafkaListenerRequest kafkaListenerRequest) {
// Existing code
}
}2. Implement Audit Logging for Listener State Changes a. Create an Audit Aspect Use Spring AOP to log listener state changes: package com.example.springbootkafkasample.aspect;
import com.example.springbootkafkasample.dto.KafkaListenerRequest;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.stereotype.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@Aspect
@Component
public class AuditAspect {
private static final Logger logger = LoggerFactory.getLogger(AuditAspect.class);
@AfterReturning(pointcut = "execution(* com.example.springbootkafkasample.service.MessageService.updateListenerState(..)) && args(request)", returning = "result")
public void logListenerStateChange(JoinPoint joinPoint, KafkaListenerRequest request, Object result) {
String containerId = request.containerId();
String operation = request.operation().name();
logger.info("Listener with containerId '{}' performed operation '{}'", containerId, operation);
}
}b. Enable AspectJ Auto-Proxy Support Make sure AspectJ proxy support is enabled: import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
@Configuration
@EnableAspectJAutoProxy
public class AppConfig {
// Other beans if necessary
}3. Implement Rate Limiting to Prevent DoS Attacks a. Add Bucket4J Dependency Include the Bucket4J library for rate limiting: <!-- Add Bucket4J dependency -->
<dependency>
<groupId>com.giffing.bucket4j.spring.boot.starter</groupId>
<artifactId>bucket4j-spring-boot-starter</artifactId>
<version>0.7.0</version>
</dependency>b. Configure Rate Limiting Create a rate-limiting filter: package com.example.springbootkafkasample.config;
import io.github.bucket4j.Bandwidth;
import io.github.bucket4j.Bucket;
import io.github.bucket4j.Refill;
import jakarta.servlet.FilterChain;
import jakarta.servlet.ServletException;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
import java.io.IOException;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Component
public class RateLimitingFilter extends OncePerRequestFilter {
private final Map<String, Bucket> cache = new ConcurrentHashMap<>();
@Override
protected void doFilterInternal(HttpServletRequest request,
HttpServletResponse response,
FilterChain filterChain) throws ServletException, IOException {
String clientIP = request.getRemoteAddr();
Bucket bucket = cache.computeIfAbsent(clientIP, this::newBucket);
if (bucket.tryConsume(1)) {
filterChain.doFilter(request, response);
} else {
response.setStatus(HttpServletResponse.SC_TOO_MANY_REQUESTS);
response.getWriter().write("Too many requests");
}
}
private Bucket newBucket(String key) {
return Bucket.builder()
.addLimit(Bandwidth.classic(5, Refill.greedy(5, Duration.ofMinutes(1))))
.build();
}
}c. Register the Filter Add the filter to your security configuration: import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.web.authentication.UsernamePasswordAuthenticationFilter;
// In your SecurityConfig class
@Autowired
private RateLimitingFilter rateLimitingFilter;
@Bean
public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {
http
// Existing configurations
.addFilterBefore(rateLimitingFilter, UsernamePasswordAuthenticationFilter.class);
return http.build();
}4. Add Request Origin Validation If your endpoint is exposed externally, validate the request origin. a. Configure CORS in Security Configuration // In your SecurityConfig class
@Bean
public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {
http
// Existing configurations
.cors().configurationSource(request -> {
CorsConfiguration config = new CorsConfiguration();
config.setAllowedOrigins(List.of("https://your-trusted-domain.com"));
config.setAllowedMethods(List.of("GET", "POST"));
return config;
})
.and()
.csrf().disable();
return http.build();
}b. Set Up Additional Origin Checks (Optional) For extra security, you can validate the // In your MessageRestController
@PostMapping
public ResponseEntity<Map<String, Boolean>> updateListenerState(
@RequestBody KafkaListenerRequest kafkaListenerRequest,
@RequestHeader(value = "Origin", required = false) String origin) {
if (origin == null || !origin.equals("https://your-trusted-domain.com")) {
return ResponseEntity.status(HttpStatus.FORBIDDEN).build();
}
// Existing code
}Note: Remember to replace placeholders like Let me know if you have any questions or need further assistance! |
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| package com.example.springbootkafkasample.dto; | ||
|
|
||
| public enum Operation { | ||
| START, | ||
| STOP | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🛠️ Refactor suggestion
Make the
GlobalExceptionHandlerclasspublicto ensure Spring can detect itThe
GlobalExceptionHandlerclass is currently package-private. To allow Spring's@ControllerAdviceto automatically detect and apply this global exception handler across the application, the class should be declared aspublic.Apply this diff to update the class access modifier:
📝 Committable suggestion