juplo.de Git - demos/kafka/training/blob - src/test/java/de/juplo/kafka/ApplicationTests.java

   1 package de.juplo.kafka;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.clients.consumer.KafkaConsumer;
   5 import org.apache.kafka.clients.producer.KafkaProducer;
   6 import org.apache.kafka.clients.producer.ProducerRecord;
   7 import org.apache.kafka.common.TopicPartition;
   8 import org.apache.kafka.common.serialization.*;
   9 import org.apache.kafka.common.utils.Bytes;
  10 import org.junit.jupiter.api.*;
  11 import org.springframework.beans.factory.annotation.Autowired;
  12 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
  13 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
  14 import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
  15 import org.springframework.boot.test.context.TestConfiguration;
  16 import org.springframework.context.annotation.Bean;
  17 import org.springframework.context.annotation.Import;
  18 import org.springframework.context.annotation.Primary;
  19 import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata;
  20 import org.springframework.kafka.support.serializer.JsonSerializer;
  21 import org.springframework.kafka.test.context.EmbeddedKafka;
  22 import org.springframework.test.context.TestPropertySource;
  23 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  24
  25 import java.time.Duration;
  26 import java.util.*;
  27 import java.util.concurrent.ExecutionException;
  28 import java.util.function.BiConsumer;
  29 import java.util.function.BiFunction;
  30 import java.util.stream.Collectors;
  31 import java.util.stream.IntStream;
  32
  33 import static de.juplo.kafka.ApplicationTests.PARTITIONS;
  34 import static de.juplo.kafka.ApplicationTests.TOPIC;
  35 import static org.assertj.core.api.Assertions.*;
  36 import static org.awaitility.Awaitility.*;
  37
  38
  39 @SpringJUnitConfig(
  40                 initializers = ConfigDataApplicationContextInitializer.class,
  41                 classes = {
  42                                 EndlessConsumer.class,
  43                                 KafkaAutoConfiguration.class,
  44                                 ApplicationTests.Configuration.class })
  45 @TestPropertySource(
  46                 properties = {
  47                                 "spring.kafka.consumer.bootstrap-servers=${spring.embedded.kafka.brokers}",
  48                                 "spring.kafka.producer.bootstrap-servers=${spring.embedded.kafka.brokers}",
  49                                 "consumer.topic=" + TOPIC })
  50 @EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS)
  51 @Slf4j
  52 class ApplicationTests
  53 {
  54         public static final String TOPIC = "FOO";
  55         public static final int PARTITIONS = 10;
  56
  57
  58         StringSerializer stringSerializer = new StringSerializer();
  59
  60         @Autowired
  61         Serializer valueSerializer;
  62         @Autowired
  63         KafkaProducer<String, Bytes> kafkaProducer;
  64         @Autowired
  65         KafkaConsumer<Bytes, Bytes> offsetConsumer;
  66         @Autowired
  67         ApplicationProperties applicationProperties;
  68         @Autowired
  69         KafkaProperties kafkaProperties;
  70         @Autowired
  71         EndlessConsumer endlessConsumer;
  72         @Autowired
  73         ClientMessageHandler clientMessageHandler;
  74
  75         Map<TopicPartition, Long> oldOffsets;
  76         Map<TopicPartition, Long> newOffsets;
  77         Set<ClientMessage> received;
  78
  79
  80         /** Tests methods */
  81
  82         @Test
  83         void commitsCurrentOffsetsOnSuccess() throws ExecutionException, InterruptedException
  84         {
  85                 send100Messages((key, counter) -> serialize(key, counter));
  86
  87                 await("100 records received")
  88                                 .atMost(Duration.ofSeconds(30))
  89                                 .until(() -> received.size() == 100);
  90
  91                 await("Offsets committed")
  92                                 .atMost(Duration.ofSeconds(10))
  93                                 .untilAsserted(() ->
  94                                 {
  95                                         checkSeenOffsetsForProgress();
  96                                         compareToCommitedOffsets(newOffsets);
  97                                 });
  98
  99                 assertThat(endlessConsumer.isRunning())
 100                                 .describedAs("Consumer should still be running")
 101                                 .isTrue();
 102         }
 103
 104         @Test
 105         void commitsCurrentOffsetsOnDeserializationError()
 106         {
 107                 send100Messages((key, counter) ->
 108                                 counter == 77
 109                                                 ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!"))
 110                                                 : serialize(key, counter));
 111
 112                 await("99 records received")
 113                                 .atMost(Duration.ofSeconds(30))
 114                                 .until(() -> received.size() == 99);
 115
 116                 await("Offsets committed")
 117                                 .atMost(Duration.ofSeconds(10))
 118                                 .untilAsserted(() ->
 119                                 {
 120                                         // UNSCHÖN:
 121                                         // Funktioniert nur, weil nach der Nachrichten, die den
 122                                         // Deserialisierungs-Fehler auslöst noch valide Nachrichten
 123                                         // gelesen werden.
 124                                         // GRUND:
 125                                         // Der MessageHandler sieht den Offset der Fehlerhaften
 126                                         // Nachricht nicht!
 127                                         checkSeenOffsetsForProgress();
 128                                         compareToCommitedOffsets(newOffsets);
 129                                 });
 130
 131                 assertThat(endlessConsumer.isRunning())
 132                                 .describedAs("Consumer should still be running")
 133                                 .isTrue();
 134         }
 135
 136         @Test
 137         void commitsOffsetOnProgramLogicErrorFoo()
 138         {
 139                 clientMessageHandler.testHandler = (clientMessage, metadata) ->
 140                 {
 141                         if (Integer.parseInt(clientMessage.message)%10 ==0)
 142                                 throw new RuntimeException("BOOM: " + clientMessage.message + "%10 == 0");
 143                 };
 144
 145                 send100Messages((key, counter) -> serialize(key, counter));
 146
 147                 await("80 records received")
 148                                 .atMost(Duration.ofSeconds(30))
 149                                 .until(() -> received.size() == 100);
 150
 151                 await("Offsets committed")
 152                                 .atMost(Duration.ofSeconds(10))
 153                                 .pollDelay(Duration.ofSeconds(1))
 154                                 .untilAsserted(() ->
 155                                 {
 156                                         checkSeenOffsetsForProgress();
 157                                         compareToCommitedOffsets(newOffsets);
 158                                 });
 159
 160                 assertThat(endlessConsumer.isRunning())
 161                                 .describedAs("Consumer should still be running")
 162                                 .isTrue();
 163         }
 164
 165
 166         /** Helper methods for the verification of expectations */
 167
 168         void compareToCommitedOffsets(Map<TopicPartition, Long> offsetsToCheck)
 169         {
 170                 doForCurrentOffsets((tp, offset) ->
 171                 {
 172                         Long expected = offsetsToCheck.get(tp) + 1;
 173                         log.debug("TEST: Comparing the expected offset of {} for {} to {}", expected, tp, offset);
 174                         assertThat(offset)
 175                                         .describedAs("Committed offset corresponds to the offset of the consumer")
 176                                         .isEqualTo(expected);
 177                 });
 178         }
 179
 180         void checkSeenOffsetsForProgress()
 181         {
 182                 // Be sure, that some messages were consumed...!
 183                 Set<TopicPartition> withProgress = new HashSet<>();
 184                 partitions().forEach(tp ->
 185                 {
 186                         Long oldOffset = oldOffsets.get(tp);
 187                         Long newOffset = newOffsets.get(tp);
 188                         if (!oldOffset.equals(newOffset))
 189                         {
 190                                 log.debug("TEST: Progress for {}: {} -> {}", tp, oldOffset, newOffset);
 191                                 withProgress.add(tp);
 192                         }
 193                 });
 194                 log.debug("TEST: Offsets with progress: {}", withProgress);
 195                 assertThat(withProgress)
 196                                 .describedAs("Some offsets must have changed, compared to the old offset-positions")
 197                                 .isNotEmpty();
 198         }
 199
 200
 201         /** Helper methods for setting up and running the tests */
 202
 203         void doForCurrentOffsets(BiConsumer<TopicPartition, Long> consumer)
 204         {
 205                 offsetConsumer.assign(partitions());
 206                 partitions().forEach(tp -> consumer.accept(tp, offsetConsumer.position(tp)));
 207                 offsetConsumer.unsubscribe();
 208         }
 209
 210         List<TopicPartition> partitions()
 211         {
 212                 return
 213                                 IntStream
 214                                                 .range(0, PARTITIONS)
 215                                                 .mapToObj(partition -> new TopicPartition(TOPIC, partition))
 216                                                 .collect(Collectors.toList());
 217         }
 218
 219
 220         void send100Messages(BiFunction<Integer, Long, Bytes> messageGenerator)
 221         {
 222                 long i = 0;
 223
 224                 for (int partition = 0; partition < 10; partition++)
 225                 {
 226                         for (int key = 0; key < 10; key++)
 227                         {
 228                                 Bytes value = messageGenerator.apply(key, ++i);
 229
 230                                 ProducerRecord<String, Bytes> record =
 231                                                 new ProducerRecord<>(
 232                                                                 TOPIC,
 233                                                                 partition,
 234                                                                 Integer.toString(key%2),
 235                                                                 value);
 236
 237                                 record.headers().add("__TypeId__", "message".getBytes());
 238                                 kafkaProducer.send(record, (metadata, e) ->
 239                                 {
 240                                         if (metadata != null)
 241                                         {
 242                                                 log.debug(
 243                                                                 "TEST: Sending partition={}, offset={} - {}={}",
 244                                                                 metadata.partition(),
 245                                                                 metadata.offset(),
 246                                                                 record.key(),
 247                                                                 record.value());
 248                                         }
 249                                         else
 250                                         {
 251                                                 log.warn(
 252                                                                 "TEST: Exception for {}={}: {}",
 253                                                                 record.key(),
 254                                                                 record.value(),
 255                                                                 e.toString());
 256                                         }
 257                                 });
 258                         }
 259                 }
 260         }
 261
 262         Bytes serialize(Integer key, Long value)
 263         {
 264                 ClientMessage message = new ClientMessage();
 265                 message.setClient(key.toString());
 266                 message.setMessage(value.toString());
 267                 return new Bytes(valueSerializer.serialize(TOPIC, message));
 268         }
 269
 270
 271         @BeforeEach
 272         public void init()
 273         {
 274                 clientMessageHandler.testHandler = (clientMessage, metadata) -> {};
 275
 276                 oldOffsets = new HashMap<>();
 277                 newOffsets = new HashMap<>();
 278                 received = new HashSet<>();
 279
 280                 doForCurrentOffsets((tp, offset) ->
 281                 {
 282                         oldOffsets.put(tp, offset - 1);
 283                         newOffsets.put(tp, offset - 1);
 284                 });
 285
 286                 clientMessageHandler.captureOffsets =
 287                                 (clientMessage, metadata) ->
 288                                 {
 289                                         received.add(clientMessage);
 290                                         log.debug("TEST: Processing record #{}: {}", received.size(), clientMessage);
 291                                         newOffsets.put(
 292                                                         new TopicPartition(metadata.topic(), metadata.partition()), metadata.offset());
 293                                 };
 294
 295                 endlessConsumer.start();
 296         }
 297
 298         @AfterEach
 299         public void deinit()
 300         {
 301                 try
 302                 {
 303                         endlessConsumer.stop();
 304                 }
 305                 catch (Exception e)
 306                 {
 307                         log.info("TEST: Exception while stopping the consumer: {}", e.toString());
 308                 }
 309         }
 310
 311         public static class ClientMessageHandler implements BiConsumer<ClientMessage, ConsumerRecordMetadata>
 312         {
 313                 BiConsumer<ClientMessage, ConsumerRecordMetadata> captureOffsets;
 314                 BiConsumer<ClientMessage, ConsumerRecordMetadata> testHandler;
 315
 316
 317                 @Override
 318                 public void accept(ClientMessage clientMessage, ConsumerRecordMetadata metadata)
 319                 {
 320                         captureOffsets
 321                                         .andThen(testHandler)
 322                                         .accept(clientMessage, metadata);
 323                 }
 324         }
 325
 326         @TestConfiguration
 327         @Import(ApplicationConfiguration.class)
 328         public static class Configuration
 329         {
 330                 @Primary
 331                 @Bean
 332                 public BiConsumer<ClientMessage, ConsumerRecordMetadata> testHandler()
 333                 {
 334                         return new ClientMessageHandler();
 335                 }
 336
 337                 @Bean
 338                 Serializer<ClientMessage> serializer()
 339                 {
 340                         return new JsonSerializer<>();
 341                 }
 342
 343                 @Bean
 344                 KafkaProducer<String, Bytes> kafkaProducer(KafkaProperties properties)
 345                 {
 346                         Properties props = new Properties();
 347                         props.put("bootstrap.servers", properties.getConsumer().getBootstrapServers());
 348                         props.put("linger.ms", 100);
 349                         props.put("key.serializer", StringSerializer.class.getName());
 350                         props.put("value.serializer", BytesSerializer.class.getName());
 351
 352                         return new KafkaProducer<>(props);
 353                 }
 354
 355                 @Bean
 356                 KafkaConsumer<Bytes, Bytes> offsetConsumer(KafkaProperties properties)
 357                 {
 358                         Properties props = new Properties();
 359                         props.put("bootstrap.servers", properties.getConsumer().getBootstrapServers());
 360                         props.put("client.id", "OFFSET-CONSUMER");
 361                         props.put("group.id", properties.getConsumer().getGroupId());
 362                         props.put("key.deserializer", BytesDeserializer.class.getName());
 363                         props.put("value.deserializer", BytesDeserializer.class.getName());
 364
 365                         return new KafkaConsumer<>(props);
 366                 }
 367         }
 368 }