-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
209 additions
and
12 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
7 changes: 7 additions & 0 deletions
7
...ter/src/main/java/com/transferwise/kafka/tkms/config/ITkmsKafkaProducerPostProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| package com.transferwise.kafka.tkms.config; | ||
|
|
||
| import java.util.function.Function; | ||
| import org.apache.kafka.clients.producer.Producer; | ||
|
|
||
| public interface ITkmsKafkaProducerPostProcessor extends Function<Producer<String, byte[]>, Producer<String, byte[]>> { | ||
| } |
12 changes: 9 additions & 3 deletions
12
...-starter/src/main/java/com/transferwise/kafka/tkms/config/ITkmsKafkaProducerProvider.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
70 changes: 70 additions & 0 deletions
70
tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/MessagePostProcessingTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| package com.transferwise.kafka.tkms; | ||
|
|
||
| import static com.transferwise.kafka.tkms.test.TestKafkaProducerPostProcessor.TEST_MESSAGE; | ||
| import static org.awaitility.Awaitility.await; | ||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| import com.transferwise.common.baseutils.transactionsmanagement.ITransactionsHelper; | ||
| import com.transferwise.kafka.tkms.api.ITransactionalKafkaMessageSender; | ||
| import com.transferwise.kafka.tkms.api.TkmsMessage; | ||
| import com.transferwise.kafka.tkms.test.BaseIntTest; | ||
| import com.transferwise.kafka.tkms.test.ITkmsSentMessagesCollector; | ||
| import com.transferwise.kafka.tkms.test.TestProperties; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.stream.StreamSupport; | ||
| import org.junit.jupiter.api.AfterEach; | ||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.junit.jupiter.api.TestInstance; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
|
|
||
| @TestInstance(TestInstance.Lifecycle.PER_CLASS) | ||
| class MessagePostProcessingTest extends BaseIntTest { | ||
|
|
||
| @Autowired | ||
| private TransactionalKafkaMessageSender transactionalKafkaMessageSender; | ||
|
|
||
| @Autowired | ||
| private TestProperties testProperties; | ||
|
|
||
| @Autowired | ||
| private ITransactionsHelper transactionsHelper; | ||
|
|
||
| @BeforeEach | ||
| void setupTest() { | ||
| tkmsSentMessagesCollector.clear(); | ||
| } | ||
|
|
||
| @AfterEach | ||
| void cleanupTest() { | ||
| tkmsSentMessagesCollector.clear(); | ||
| } | ||
|
|
||
| @Test | ||
| void messagesAreInstrumentedWithProducerPostProcessor() { | ||
| byte[] someValue = TEST_MESSAGE; | ||
|
|
||
| String topic = testProperties.getTestTopic(); | ||
|
|
||
| transactionsHelper.withTransaction().run(() -> | ||
| transactionalKafkaMessageSender.sendMessages(new ITransactionalKafkaMessageSender.SendMessagesRequest() | ||
| .addTkmsMessage(new TkmsMessage().setTopic(topic).setKey("1").setValue(someValue)) | ||
| .addTkmsMessage(new TkmsMessage().setTopic(topic).setKey("2").setValue(someValue)) | ||
| )); | ||
|
|
||
| await().until(() -> tkmsSentMessagesCollector.getSentMessages(topic).size() == 2); | ||
| var messages = tkmsSentMessagesCollector.getSentMessages(topic); | ||
|
|
||
| assertEquals(2, messages.size()); | ||
| checkForHeader(messages.get(0), "wrapTest", "wrapped"); | ||
| checkForHeader(messages.get(1), "wrapTest", "wrapped"); | ||
| } | ||
|
|
||
| private void checkForHeader(ITkmsSentMessagesCollector.SentMessage sentMessage, String key, String value) { | ||
| assertTrue( | ||
| StreamSupport.stream(sentMessage.getProducerRecord().headers().spliterator(), false) | ||
| .anyMatch(h -> h.key().equals(key) && value.equals(new String(h.value(), StandardCharsets.UTF_8))) | ||
| ); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
76 changes: 76 additions & 0 deletions
76
...tarter/src/test/java/com/transferwise/kafka/tkms/test/TestKafkaProducerPostProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| package com.transferwise.kafka.tkms.test; | ||
|
|
||
| import com.transferwise.kafka.tkms.config.ITkmsKafkaProducerPostProcessor; | ||
| import com.transferwise.kafka.tkms.config.TkmsKafkaProducerProvider; | ||
| import java.lang.reflect.InvocationHandler; | ||
| import java.lang.reflect.InvocationTargetException; | ||
| import java.lang.reflect.Method; | ||
| import java.lang.reflect.Proxy; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Arrays; | ||
| import org.apache.kafka.clients.producer.Callback; | ||
| import org.apache.kafka.clients.producer.Producer; | ||
| import org.apache.kafka.clients.producer.ProducerRecord; | ||
| import org.springframework.beans.factory.InitializingBean; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| @Component | ||
| public class TestKafkaProducerPostProcessor implements ITkmsKafkaProducerPostProcessor, InitializingBean { | ||
|
|
||
| public static final byte[] TEST_MESSAGE = "Testing ProducerPostProcessing".getBytes(StandardCharsets.UTF_8); | ||
|
|
||
| private ProxyInvocationHandler handler; | ||
|
|
||
| @Autowired | ||
| TkmsKafkaProducerProvider tkmsKafkaProducerProvider; | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| @Override | ||
| public Producer<String, byte[]> apply(Producer<String, byte[]> producer) { | ||
| handler = new ProxyInvocationHandler(producer); | ||
| return (Producer<String, byte[]>) | ||
| Proxy.newProxyInstance( | ||
| TestKafkaProducerPostProcessor.class.getClassLoader(), | ||
| new Class<?>[] {Producer.class}, | ||
| handler); | ||
| } | ||
|
|
||
| @Override | ||
| public void afterPropertiesSet() throws Exception { | ||
| tkmsKafkaProducerProvider.addPostProcessor(this); | ||
| } | ||
|
|
||
| private static class ProxyInvocationHandler implements InvocationHandler { | ||
|
|
||
| private final Producer<String, byte[]> producer; | ||
|
|
||
| public ProxyInvocationHandler(Producer<String, byte[]> producer) { | ||
| this.producer = producer; | ||
| } | ||
|
|
||
| @Override | ||
| public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { | ||
| if ("send".equals(method.getName()) | ||
| && method.getParameterCount() >= 1 | ||
| && method.getParameterTypes()[0] == ProducerRecord.class) { | ||
| ProducerRecord<String, byte[]> record = (ProducerRecord<String, byte[]>) args[0]; | ||
| if (Arrays.equals(TEST_MESSAGE, record.value())) { | ||
| record.headers().add("wrapTest", "wrapped".getBytes(StandardCharsets.UTF_8)); | ||
| } | ||
| Callback callback = | ||
| method.getParameterCount() >= 2 | ||
| && method.getParameterTypes()[1] == Callback.class | ||
| ? (Callback) args[1] | ||
| : null; | ||
| return producer.send(record, callback); | ||
| } else { | ||
| try { | ||
| return method.invoke(producer, args); | ||
| } catch (InvocationTargetException exception) { | ||
| throw exception.getCause(); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |