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

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