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

   1 package de.juplo.kafka;
   2
   3 import com.mongodb.client.MongoClient;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.apache.kafka.clients.consumer.ConsumerConfig;
   6 import org.apache.kafka.clients.consumer.KafkaConsumer;
   7 import org.apache.kafka.clients.producer.KafkaProducer;
   8 import org.apache.kafka.clients.producer.ProducerRecord;
   9 import org.apache.kafka.common.TopicPartition;
  10 import org.apache.kafka.common.serialization.*;
  11 import org.apache.kafka.common.utils.Bytes;
  12 import org.junit.jupiter.api.*;
  13 import org.springframework.beans.factory.annotation.Autowired;
  14 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
  15 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
  16 import org.springframework.boot.autoconfigure.mongo.MongoProperties;
  17 import org.springframework.boot.test.autoconfigure.data.mongo.AutoConfigureDataMongo;
  18 import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
  19 import org.springframework.boot.test.context.TestConfiguration;
  20 import org.springframework.context.annotation.Bean;
  21 import org.springframework.context.annotation.Import;
  22 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  23 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
  24 import org.springframework.kafka.core.ConsumerFactory;
  25 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  26 import org.springframework.kafka.test.context.EmbeddedKafka;
  27 import org.springframework.test.context.TestPropertySource;
  28 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  29
  30 import java.time.Duration;
  31 import java.util.*;
  32 import java.util.function.BiConsumer;
  33 import java.util.function.Consumer;
  34 import java.util.stream.Collectors;
  35 import java.util.stream.IntStream;
  36
  37 import static de.juplo.kafka.GenericApplicationTests.PARTITIONS;
  38 import static de.juplo.kafka.GenericApplicationTests.TOPIC;
  39 import static org.assertj.core.api.Assertions.*;
  40 import static org.awaitility.Awaitility.*;
  41
  42
  43 @SpringJUnitConfig(initializers = ConfigDataApplicationContextInitializer.class)
  44 @TestPropertySource(
  45                 properties = {
  46                                 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  47                                 "spring.kafka.producer.bootstrap-servers=${spring.embedded.kafka.brokers}",
  48                                 "sumup.adder.topic=" + TOPIC,
  49                                 "spring.kafka.consumer.auto-commit-interval=500ms",
  50                                 "spring.mongodb.embedded.version=4.4.13" })
  51 @EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS)
  52 @EnableAutoConfiguration
  53 @AutoConfigureDataMongo
  54 @Slf4j
  55 abstract class GenericApplicationTests<K, V>
  56 {
  57         public static final String TOPIC = "FOO";
  58         public static final int PARTITIONS = 10;
  59
  60
  61         @Autowired
  62         org.apache.kafka.clients.consumer.Consumer<K, V> kafkaConsumer;
  63         @Autowired
  64         KafkaProperties kafkaProperties;
  65         @Autowired
  66         ApplicationProperties applicationProperties;
  67         @Autowired
  68         MongoClient mongoClient;
  69         @Autowired
  70         MongoProperties mongoProperties;
  71         @Autowired
  72         KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
  73         @Autowired
  74         TestRecordHandler recordHandler;
  75         @Autowired
  76         DeadLetterTopicConsumer deadLetterTopicConsumer;
  77         @Autowired
  78         EndlessConsumer endlessConsumer;
  79
  80         KafkaProducer<Bytes, Bytes> testRecordProducer;
  81         KafkaConsumer<Bytes, Bytes> offsetConsumer;
  82         Map<TopicPartition, Long> oldOffsets;
  83
  84
  85         final RecordGenerator recordGenerator;
  86         final Consumer<ProducerRecord<Bytes, Bytes>> messageSender;
  87
  88         public GenericApplicationTests(RecordGenerator recordGenerator)
  89         {
  90                 this.recordGenerator = recordGenerator;
  91                 this.messageSender = (record) -> sendMessage(record);
  92         }
  93
  94
  95         /** Tests methods */
  96
  97         @Test
  98         void commitsCurrentOffsetsOnSuccess() throws Exception
  99         {
 100                 recordGenerator.generate(false, false, messageSender);
 101
 102                 int numberOfGeneratedMessages = recordGenerator.getNumberOfMessages();
 103
 104                 await(numberOfGeneratedMessages + " records received")
 105                                 .atMost(Duration.ofSeconds(30))
 106                                 .pollInterval(Duration.ofSeconds(1))
 107                                 .until(() -> recordHandler.receivedMessages >= numberOfGeneratedMessages);
 108
 109                 await("Offsets committed")
 110                                 .atMost(Duration.ofSeconds(10))
 111                                 .pollInterval(Duration.ofSeconds(1))
 112                                 .untilAsserted(() ->
 113                                 {
 114                                         checkSeenOffsetsForProgress();
 115                                         assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
 116                                 });
 117
 118                 assertThat(endlessConsumer.running())
 119                                 .describedAs("Consumer should still be running")
 120                                 .isTrue();
 121
 122                 endlessConsumer.stop();
 123                 recordGenerator.assertBusinessLogic();
 124         }
 125
 126         @Test
 127         @SkipWhenErrorCannotBeGenerated(poisonPill = true)
 128         void commitsOffsetOfErrorForReprocessingOnDeserializationError()
 129         {
 130                 recordGenerator.generate(true, false, messageSender);
 131
 132                 int numberOfValidMessages =
 133                                 recordGenerator.getNumberOfMessages() -
 134                                 recordGenerator.getNumberOfPoisonPills();
 135
 136                 await(numberOfValidMessages + " records received")
 137                                 .atMost(Duration.ofSeconds(30))
 138                                 .pollInterval(Duration.ofSeconds(1))
 139                                 .until(() -> recordHandler.receivedMessages >= numberOfValidMessages);
 140                 await(recordGenerator.getNumberOfPoisonPills() + " poison-pills received")
 141                                 .atMost(Duration.ofSeconds(30))
 142                                 .pollInterval(Duration.ofSeconds(1))
 143                                 .until(() -> deadLetterTopicConsumer.messages.size() == recordGenerator.getNumberOfPoisonPills());
 144
 145                 await("Offsets committed")
 146                                 .atMost(Duration.ofSeconds(10))
 147                                 .pollInterval(Duration.ofSeconds(1))
 148                                 .untilAsserted(() ->
 149                                 {
 150                                         checkSeenOffsetsForProgress();
 151                                         assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
 152                                 });
 153
 154                 assertThat(endlessConsumer.running())
 155                                 .describedAs("Consumer should still be running")
 156                                 .isTrue();
 157
 158                 endlessConsumer.stop();
 159                 recordGenerator.assertBusinessLogic();
 160         }
 161
 162         @Test
 163         @SkipWhenErrorCannotBeGenerated(logicError = true)
 164         void commitsOffsetsOfUnseenRecordsOnLogicError()
 165         {
 166                 recordGenerator.generate(false, true, messageSender);
 167
 168                 int numberOfValidMessages =
 169                                 recordGenerator.getNumberOfMessages() -
 170                                 recordGenerator.getNumberOfLogicErrors();
 171
 172                 await(numberOfValidMessages + " records received")
 173                                 .atMost(Duration.ofSeconds(30))
 174                                 .pollInterval(Duration.ofSeconds(1))
 175                                 .until(() -> recordHandler.receivedMessages >= numberOfValidMessages);
 176                 await(recordGenerator.getNumberOfLogicErrors() + " logic-errors received")
 177                                 .atMost(Duration.ofSeconds(30))
 178                                 .pollInterval(Duration.ofSeconds(1))
 179                                 .until(() -> deadLetterTopicConsumer.messages.size() == recordGenerator.getNumberOfLogicErrors());
 180
 181                 await("Offsets committed")
 182                                 .atMost(Duration.ofSeconds(10))
 183                                 .pollInterval(Duration.ofSeconds(1))
 184                                 .untilAsserted(() ->
 185                                 {
 186                                         checkSeenOffsetsForProgress();
 187                                         assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
 188                                 });
 189
 190                 assertThat(endlessConsumer.running())
 191                                 .describedAs("Consumer should still be running")
 192                                 .isTrue();
 193
 194                 endlessConsumer.stop();
 195                 recordGenerator.assertBusinessLogic();
 196         }
 197
 198
 199         /** Helper methods for the verification of expectations */
 200
 201         void assertSeenOffsetsEqualCommittedOffsets(Map<TopicPartition, Long> offsetsToCheck)
 202         {
 203                 doForCurrentOffsets((tp, offset) ->
 204                 {
 205                         Long expected = offsetsToCheck.get(tp) + 1;
 206                         log.debug("Checking, if the offset {} for {} is exactly {}", offset, tp, expected);
 207                         assertThat(offset)
 208                                         .describedAs("Committed offset corresponds to the offset of the consumer")
 209                                         .isEqualTo(expected);
 210                 });
 211         }
 212
 213         void assertSeenOffsetsAreBehindCommittedOffsets(Map<TopicPartition, Long> offsetsToCheck)
 214         {
 215                 List<Boolean> isOffsetBehindSeen = new LinkedList<>();
 216
 217                 doForCurrentOffsets((tp, offset) ->
 218                 {
 219                         Long expected = offsetsToCheck.get(tp) + 1;
 220                         log.debug("Checking, if the offset {} for {} is at most {}", offset, tp, expected);
 221                         assertThat(offset)
 222                                         .describedAs("Committed offset must be at most equal to the offset of the consumer")
 223                                         .isLessThanOrEqualTo(expected);
 224                         isOffsetBehindSeen.add(offset < expected);
 225                 });
 226
 227                 assertThat(isOffsetBehindSeen.stream().reduce(false, (result, next) -> result | next))
 228                                 .describedAs("Committed offsets are behind seen offsets")
 229                                 .isTrue();
 230         }
 231
 232         void checkSeenOffsetsForProgress()
 233         {
 234                 // Be sure, that some messages were consumed...!
 235                 Set<TopicPartition> withProgress = new HashSet<>();
 236                 partitions().forEach(tp ->
 237                 {
 238                         Long oldOffset = oldOffsets.get(tp) + 1;
 239                         Long newOffset = recordHandler.seenOffsets.get(tp) + 1;
 240                         if (!oldOffset.equals(newOffset))
 241                         {
 242                                 log.debug("Progress for {}: {} -> {}", tp, oldOffset, newOffset);
 243                                 withProgress.add(tp);
 244                         }
 245                 });
 246                 assertThat(withProgress)
 247                                 .describedAs("Some offsets must have changed, compared to the old offset-positions")
 248                                 .isNotEmpty();
 249         }
 250
 251
 252         /** Helper methods for setting up and running the tests */
 253
 254         void seekToEnd()
 255         {
 256                 offsetConsumer.assign(partitions());
 257                 offsetConsumer.seekToEnd(partitions());
 258                 partitions().forEach(tp ->
 259                 {
 260                         // seekToEnd() works lazily: it only takes effect on poll()/position()
 261                         Long offset = offsetConsumer.position(tp);
 262                         log.info("New position for {}: {}", tp, offset);
 263                 });
 264                 // The new positions must be commited!
 265                 offsetConsumer.commitSync();
 266                 offsetConsumer.unsubscribe();
 267         }
 268
 269         void doForCurrentOffsets(BiConsumer<TopicPartition, Long> consumer)
 270         {
 271                 offsetConsumer.assign(partitions());
 272                 partitions().forEach(tp -> consumer.accept(tp, offsetConsumer.position(tp)));
 273                 offsetConsumer.unsubscribe();
 274         }
 275
 276         List<TopicPartition> partitions()
 277         {
 278                 return
 279                                 IntStream
 280                                                 .range(0, PARTITIONS)
 281                                                 .mapToObj(partition -> new TopicPartition(TOPIC, partition))
 282                                                 .collect(Collectors.toList());
 283         }
 284
 285
 286         public interface RecordGenerator
 287         {
 288                 void generate(
 289                                 boolean poisonPills,
 290                                 boolean logicErrors,
 291                                 Consumer<ProducerRecord<Bytes, Bytes>> messageSender);
 292
 293                 int getNumberOfMessages();
 294                 int getNumberOfPoisonPills();
 295                 int getNumberOfLogicErrors();
 296
 297                 default boolean canGeneratePoisonPill()
 298                 {
 299                         return true;
 300                 }
 301
 302                 default boolean canGenerateLogicError()
 303                 {
 304                         return true;
 305                 }
 306
 307                 default void assertBusinessLogic()
 308                 {
 309                         log.debug("No business-logic to assert");
 310                 }
 311         }
 312
 313         void sendMessage(ProducerRecord<Bytes, Bytes> record)
 314         {
 315                 testRecordProducer.send(record, (metadata, e) ->
 316                 {
 317                         if (metadata != null)
 318                         {
 319                                 log.debug(
 320                                                 "{}|{} - {}={}",
 321                                                 metadata.partition(),
 322                                                 metadata.offset(),
 323                                                 record.key(),
 324                                                 record.value());
 325                         }
 326                         else
 327                         {
 328                                 log.warn(
 329                                                 "Exception for {}={}: {}",
 330                                                 record.key(),
 331                                                 record.value(),
 332                                                 e.toString());
 333                         }
 334                 });
 335         }
 336
 337
 338         @BeforeEach
 339         public void init()
 340         {
 341                 Properties props;
 342                 props = new Properties();
 343                 props.put("bootstrap.servers", kafkaProperties.getBootstrapServers());
 344                 props.put("linger.ms", 100);
 345                 props.put("key.serializer", BytesSerializer.class.getName());
 346                 props.put("value.serializer", BytesSerializer.class.getName());
 347                 testRecordProducer = new KafkaProducer<>(props);
 348
 349                 props = new Properties();
 350                 props.put("bootstrap.servers", kafkaProperties.getBootstrapServers());
 351                 props.put("client.id", "OFFSET-CONSUMER");
 352                 props.put("group.id", kafkaProperties.getConsumer().getGroupId());
 353                 props.put("key.deserializer", BytesDeserializer.class.getName());
 354                 props.put("value.deserializer", BytesDeserializer.class.getName());
 355                 offsetConsumer = new KafkaConsumer<>(props);
 356
 357                 mongoClient.getDatabase(mongoProperties.getDatabase()).drop();
 358                 seekToEnd();
 359
 360                 oldOffsets = new HashMap<>();
 361                 recordHandler.seenOffsets = new HashMap<>();
 362                 recordHandler.receivedMessages = 0;
 363
 364                 deadLetterTopicConsumer.messages.clear();
 365
 366                 doForCurrentOffsets((tp, offset) ->
 367                 {
 368                         oldOffsets.put(tp, offset - 1);
 369                         recordHandler.seenOffsets.put(tp, offset - 1);
 370                 });
 371
 372                 endlessConsumer.start();
 373         }
 374
 375         @AfterEach
 376         public void deinit()
 377         {
 378                 try
 379                 {
 380                         endlessConsumer.stop();
 381                 }
 382                 catch (Exception e)
 383                 {
 384                         log.debug("{}", e.toString());
 385                 }
 386
 387                 try
 388                 {
 389                         testRecordProducer.close();
 390                         offsetConsumer.close();
 391                 }
 392                 catch (Exception e)
 393                 {
 394                         log.info("Exception while stopping the consumer: {}", e.toString());
 395                 }
 396         }
 397
 398
 399         @TestConfiguration
 400         @Import(ApplicationConfiguration.class)
 401         public static class Configuration
 402         {
 403                 @Bean
 404                 public RecordHandler recordHandler(RecordHandler applicationRecordHandler)
 405                 {
 406                         return new TestRecordHandler(applicationRecordHandler);
 407                 }
 408
 409                 @Bean(destroyMethod = "close")
 410                 public org.apache.kafka.clients.consumer.Consumer<String, Message> kafkaConsumer(ConsumerFactory<String, Message> factory)
 411                 {
 412                         return factory.createConsumer();
 413                 }
 414
 415     @Bean
 416     public ConcurrentKafkaListenerContainerFactory<String, String> dltContainerFactory(
 417       KafkaProperties properties)
 418     {
 419       Map<String, Object> consumerProperties = new HashMap<>();
 420
 421       consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
 422       consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 423       consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 424       consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 425
 426       DefaultKafkaConsumerFactory dltConsumerFactory =
 427         new DefaultKafkaConsumerFactory<>(consumerProperties);
 428       ConcurrentKafkaListenerContainerFactory<String, String> factory =
 429         new ConcurrentKafkaListenerContainerFactory<>();
 430       factory.setConsumerFactory(dltConsumerFactory);
 431       return factory;
 432     }
 433
 434                 @Bean
 435                 public DeadLetterTopicConsumer deadLetterTopicConsumer()
 436                 {
 437                         return new DeadLetterTopicConsumer();
 438                 }
 439         }
 440 }