Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add ability to exclude/include kafka event detail on producer and con… #47

Open
wants to merge 2 commits into
base: feature/spring-boot-2.7.0
Choose a base branch
from
Open

add ability to exclude/include kafka event detail on producer and con… #47

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2b73161
add ability to exclude/include kafka event detail on producer and con…
Feb 16, 2023
90835c1
update readme
Feb 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions blibli-backend-framework-kafka/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,4 +136,12 @@ blibli.backend.kafka.logging.before-send=true
blibli.backend.kafka.logging.before-consume=true
blibli.backend.kafka.logging.after-success-consume=true
blibli.backend.kafka.logging.after-failed-consume=true
```

By default if the log interceptor is turned on it will show the payload on each log, we can make it excluded from the log to save log spaces by setting these properties
```properties
blibli.backend.kafka.logging.before-send-exclude-event=true
blibli.backend.kafka.logging.before-consume-exclude-event=true
blibli.backend.kafka.logging.after-success-consume-exclude-event=true
blibli.backend.kafka.logging.after-failed-consume-exclude-event=true
```
18 changes: 15 additions & 3 deletions ...c/main/java/com/blibli/oss/backend/kafka/interceptor/log/LogKafkaConsumerInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,22 +20,34 @@ public class LogKafkaConsumerInterceptor implements KafkaConsumerInterceptor, Or
@Override
public boolean beforeConsume(ConsumerRecord<String, String> consumerRecord) {
if (kafkaProperties.getLogging().isBeforeConsume()) {
log.info("Receive from topic {} with message {}:{}", consumerRecord.topic(), consumerRecord.key(), consumerRecord.value());
if (kafkaProperties.getLogging().isBeforeConsumeExcludeEvent()) {
log.info("Receive from topic {} with message key: {}", consumerRecord.topic(), consumerRecord.key());
} else {
log.info("Receive from topic {} with message {}:{}", consumerRecord.topic(), consumerRecord.key(), consumerRecord.value());
}
}
return false;
}

@Override
public void afterSuccessConsume(ConsumerRecord<String, String> consumerRecord) {
if (kafkaProperties.getLogging().isAfterSuccessConsume()) {
log.info("Success consume from topic {} with message {}:{}", consumerRecord.topic(), consumerRecord.key(), consumerRecord.value());
if (kafkaProperties.getLogging().isAfterSuccessExcludeEvent()) {
log.info("Success consume from topic {} with message key: {}", consumerRecord.topic(), consumerRecord.key());
} else {
log.info("Success consume from topic {} with message {}:{}", consumerRecord.topic(), consumerRecord.key(), consumerRecord.value());
}
}
}

@Override
public void afterFailedConsume(ConsumerRecord<String, String> consumerRecord, Throwable throwable) {
if (kafkaProperties.getLogging().isAfterFailedConsume()) {
log.error(String.format("Failed consume from topic %s with message %s:%s and exception %s", consumerRecord.topic(), consumerRecord.key(), consumerRecord.value(), throwable.getMessage()), throwable);
if (kafkaProperties.getLogging().isAfterFailedExcludeEvent()) {
log.error(String.format("Failed consume from topic %s with message key: %s and exception %s", consumerRecord.topic(), consumerRecord.key(), throwable.getMessage()), throwable);
} else {
log.error(String.format("Failed consume from topic %s with message %s:%s and exception %s", consumerRecord.topic(), consumerRecord.key(), consumerRecord.value(), throwable.getMessage()), throwable);
}
}
}
}
6 changes: 5 additions & 1 deletion ...c/main/java/com/blibli/oss/backend/kafka/interceptor/log/LogKafkaProducerInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,11 @@ public class LogKafkaProducerInterceptor implements KafkaProducerInterceptor, Or
@Override
public ProducerEvent beforeSend(ProducerEvent event) {
if (kafkaProperties.getLogging().isBeforeSend()) {
log.info("Send message to kafka : {}", event);
if (kafkaProperties.getLogging().isBeforeSendExcludeEvent()) {
log.info("Send message to kafka : {topic: {}, key: {}}", event.getTopic(), event.getKey());
} else {
log.info("Send message to kafka : {}", event);
}
}
return event;
}
Expand Down
8 changes: 8 additions & 0 deletions ...ramework-kafka/src/main/java/com/blibli/oss/backend/kafka/properties/KafkaProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,14 @@ public static class LoggingProperties {

private boolean afterFailedConsume = false;

private boolean beforeSendExcludeEvent = false;

private boolean beforeConsumeExcludeEvent = false;

private boolean afterSuccessExcludeEvent = false;

private boolean afterFailedExcludeEvent = false;

}

}
86 changes: 81 additions & 5 deletions .../java/com/blibli/oss/backend/kafka/interceptor/consumer/KafkaConsumerInterceptorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import com.blibli.oss.backend.kafka.interceptor.KafkaConsumerInterceptor;
import com.blibli.oss.backend.kafka.producer.KafkaProducer;
import com.blibli.oss.backend.kafka.producer.helper.KafkaHelper;
import com.blibli.oss.backend.kafka.properties.KafkaProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
Expand All @@ -13,6 +14,8 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.system.CapturedOutput;
import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
Expand All @@ -24,7 +27,9 @@
import java.lang.reflect.Method;
import java.util.Collections;

@ExtendWith(SpringExtension.class)
import static org.junit.jupiter.api.Assertions.assertTrue;

@ExtendWith({SpringExtension.class, OutputCaptureExtension.class})
@SpringBootTest(classes = KafkaConsumerInterceptorTest.Application.class)
@EmbeddedKafka(
partitions = 1,
Expand Down Expand Up @@ -53,6 +58,9 @@ class KafkaConsumerInterceptorTest {
@Autowired
private EmbeddedKafkaBroker broker;

@Autowired
private KafkaProperties kafkaProperties;

private Consumer<String, String> consumer;

@BeforeEach
Expand All @@ -62,6 +70,10 @@ void setUp() {

helloInterceptor.reset();
counterInterceptor.reset();

kafkaProperties.getLogging().setBeforeConsumeExcludeEvent(false);
kafkaProperties.getLogging().setAfterSuccessExcludeEvent(false);
kafkaProperties.getLogging().setAfterFailedExcludeEvent(false);
}

@AfterEach
Expand All @@ -70,7 +82,7 @@ void tearDown() {
}

@Test
@Order(4)
@Order(6)
void testListener() throws InterruptedException {
kafkaProducer.sendAndSubscribe(TOPIC, "key", "value", Schedulers.boundedElastic());
Thread.sleep(2_000L); // wait 5 seconds until message received by listener
Expand All @@ -88,25 +100,89 @@ void testListener() throws InterruptedException {
}

@Test
@Order(3)
void testListenerWithInterceptor() throws InterruptedException {
@Order(4)
void testListenerWithInterceptor(CapturedOutput output) throws InterruptedException {
kafkaProducer.sendAndSubscribe(TOPIC_GOODBYE, "key", "value", Schedulers.boundedElastic());
Thread.sleep(4_000L); // wait 5 seconds until message received by listener

Assertions.assertEquals(1, counterInterceptor.getBeforeConsume());
Assertions.assertEquals(1, counterInterceptor.getAfterSuccessConsume());

assertTrue(
output.getOut()
.contains("Receive from topic " + TOPIC_GOODBYE + " with message key:value")
);
assertTrue(
output.getOut()
.contains("Success consume from topic " + TOPIC_GOODBYE + " with message key:value")
);
}

@Test
@Order(5)
void testListenerWithInterceptorExcludeEventOnLogs(CapturedOutput output) throws InterruptedException {
kafkaProperties.getLogging().setBeforeConsumeExcludeEvent(true);
kafkaProperties.getLogging().setAfterSuccessExcludeEvent(true);

kafkaProducer.sendAndSubscribe(TOPIC_GOODBYE, "key", "value", Schedulers.boundedElastic());
Thread.sleep(4_000L); // wait 5 seconds until message received by listener

Assertions.assertEquals(1, counterInterceptor.getBeforeConsume());
Assertions.assertEquals(1, counterInterceptor.getAfterSuccessConsume());

assertTrue(
output.getOut()
.contains("Receive from topic " + TOPIC_GOODBYE + " with message key: key")
);
assertTrue(
output.getOut()
.contains("Success consume from topic " + TOPIC_GOODBYE + " with message key: key")
);
}

@Test
@Order(2)
void testInterceptorError() throws InterruptedException {
void testInterceptorError(CapturedOutput output) throws InterruptedException {
kafkaProducer.sendAndSubscribe(TOPIC, "error", "value", Schedulers.boundedElastic());
Thread.sleep(2_000L); // wait 5 seconds until message received by listener

Assertions.assertEquals(helloInterceptor.beforeConsume, "value");
Assertions.assertEquals(helloInterceptor.afterFailed, "value");
Assertions.assertEquals(0, counterInterceptor.getBeforeConsume());
Assertions.assertEquals(0, counterInterceptor.getAfterSuccessConsume());

assertTrue(
output.getOut()
.contains("Receive from topic " + TOPIC + " with message error:value")
);
assertTrue(
output.getOut()
.contains("Failed consume from topic " + TOPIC + " with message error:value")
);
}

@Test
@Order(3)
void testInterceptorErrorExcludeEventOnLogs(CapturedOutput output) throws InterruptedException {
kafkaProperties.getLogging().setBeforeConsumeExcludeEvent(true);
kafkaProperties.getLogging().setAfterFailedExcludeEvent(true);

kafkaProducer.sendAndSubscribe(TOPIC, "error", "value", Schedulers.boundedElastic());
Thread.sleep(2_000L); // wait 5 seconds until message received by listener

Assertions.assertEquals(helloInterceptor.beforeConsume, "value");
Assertions.assertEquals(helloInterceptor.afterFailed, "value");
Assertions.assertEquals(0, counterInterceptor.getBeforeConsume());
Assertions.assertEquals(0, counterInterceptor.getAfterSuccessConsume());

assertTrue(
output.getOut()
.contains("Receive from topic " + TOPIC + " with message key: error")
);
assertTrue(
output.getOut()
.contains("Failed consume from topic " + TOPIC + " with message key: error")
);
}

@Test
Expand Down
37 changes: 35 additions & 2 deletions .../java/com/blibli/oss/backend/kafka/interceptor/producer/KafkaProducerInterceptorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import com.blibli.oss.backend.kafka.model.ProducerEvent;
import com.blibli.oss.backend.kafka.producer.KafkaProducer;
import com.blibli.oss.backend.kafka.producer.helper.KafkaHelper;
import com.blibli.oss.backend.kafka.properties.KafkaProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.Data;
Expand All @@ -17,6 +18,8 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.system.CapturedOutput;
import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
Expand All @@ -30,7 +33,7 @@

import static org.junit.jupiter.api.Assertions.*;

@ExtendWith(SpringExtension.class)
@ExtendWith({SpringExtension.class, OutputCaptureExtension.class})
@SpringBootTest(classes = KafkaProducerInterceptorTest.Application.class)
@EmbeddedKafka(
topics = KafkaProducerInterceptorTest.TOPIC
Expand All @@ -51,10 +54,15 @@ class KafkaProducerInterceptorTest {
@Autowired
private ErrorFlag errorFlag;

@Autowired
private KafkaProperties kafkaProperties;

@BeforeEach
void setUp() {
consumer = KafkaHelper.newConsumer(broker);
consumer.subscribe(Collections.singletonList(TOPIC));

kafkaProperties.getLogging().setBeforeConsumeExcludeEvent(false);
}

@AfterEach
Expand All @@ -63,14 +71,39 @@ void tearDown() {
}

@Test
void testInterceptor() {
void testInterceptor(CapturedOutput output) {
errorFlag.setError(false);
kafkaProducer.sendAndSubscribe(TOPIC, "key", "value", Schedulers.boundedElastic());

ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, TOPIC);

Assertions.assertEquals(record.key(), "key");
Assertions.assertEquals(record.value(), "value changed");

assertTrue(
output.getOut()
.contains("Send message to kafka : " +
"ProducerEvent(topic=KafkaProducerInterceptorTest, partition=null, headers=null, key=key, value=value, timestamp=null)")
);
}

@Test
void testInterceptorExcludeEventOnLogs(CapturedOutput output) {
kafkaProperties.getLogging().setBeforeSendExcludeEvent(true);

errorFlag.setError(false);
kafkaProducer.sendAndSubscribe(TOPIC, "key", "value", Schedulers.boundedElastic());

ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, TOPIC);

Assertions.assertEquals(record.key(), "key");
Assertions.assertEquals(record.value(), "value changed");

assertTrue(
output.getOut()
.contains("Send message to kafka : " +
"{topic: KafkaProducerInterceptorTest, key: key}")
);
}

@Test
Expand Down