From cac7d654082dad21c90e3e9ec2b6cfefa4961076 Mon Sep 17 00:00:00 2001 From: aalehin Date: Mon, 12 Sep 2022 13:37:54 +0300 Subject: [PATCH] http://git.mfd.msk/dashboard/issues?assignee_id=30 --- small refactor --- .../service/BankAccountServiceTest.java | 144 +----------------- 1 file changed, 4 insertions(+), 140 deletions(-) diff --git a/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java index 907a76f6e..621c5e9e4 100644 --- a/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java +++ b/clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/BankAccountServiceTest.java @@ -1,8 +1,9 @@ package ru.spcex.clearing.account.service; -import org.apache.kafka.clients.consumer.*; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.MockConsumer; +import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.errors.WakeupException; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -13,35 +14,20 @@ import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgHazelcast; -import java.time.Duration; -import java.time.temporal.ChronoUnit; -import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; -import java.util.List; -import java.util.stream.StreamSupport; -import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat; import static org.mockito.Mockito.doReturn; class BankAccountServiceTest { private static final int PARTITION = 0; private static final String TOPIC = Consts.DESTINATION_BANK_ACCOUNT_NEW; - private MockConsumer consumer; private MockConsumer mockConsumer; - private List updates; - private CountryPopulationConsumer countryPopulationConsumer; - private Throwable pollException; @BeforeEach void setUp() { mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST); - - consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST); - updates = new ArrayList<>(); - countryPopulationConsumer = new CountryPopulationConsumer(consumer, - ex -> this.pollException = ex, updates::add); } @Test @@ -49,7 +35,7 @@ class BankAccountServiceTest { //arrange mockConsumer.schedulePollTask(() -> { mockConsumer.rebalance(Collections.singletonList(new TopicPartition(TOPIC, PARTITION))); - mockConsumer.addRecord(new ConsumerRecord<>(TOPIC, PARTITION, 0, "key", "value")); + mockConsumer.addRecord(new ConsumerRecord<>(TOPIC, PARTITION, 0, "key", "test-value")); }); mockConsumer.schedulePollTask(() -> mockConsumer.wakeup()); @@ -78,126 +64,4 @@ class BankAccountServiceTest { //assert } - - @Test - void whenStartingByAssigningTopicPartition_thenExpectUpdatesAreConsumedCorrectly() { - // GIVEN - consumer.schedulePollTask(() -> consumer.addRecord(record(TOPIC, PARTITION, "Romania", 19_410_000))); - consumer.schedulePollTask(() -> countryPopulationConsumer.stop()); - - HashMap startOffsets = new HashMap<>(); - TopicPartition tp = new TopicPartition(TOPIC, PARTITION); - startOffsets.put(tp, 0L); - consumer.updateBeginningOffsets(startOffsets); - - // WHEN - countryPopulationConsumer.startByAssigning(TOPIC, PARTITION); - - // THEN - assertThat(updates).hasSize(1); - assertThat(consumer.closed()).isTrue(); - } - - @Test - void whenStartingBySubscribingToTopic_thenExpectUpdatesAreConsumedCorrectly() { - // GIVEN - consumer.schedulePollTask(() -> { - consumer.rebalance(Collections.singletonList(new TopicPartition(TOPIC, PARTITION))); - consumer.addRecord(record(TOPIC, PARTITION, "Romania", 20)); - }); - consumer.schedulePollTask(() -> countryPopulationConsumer.stop()); - - HashMap startOffsets = new HashMap<>(); - TopicPartition tp = new TopicPartition(TOPIC, 0); - startOffsets.put(tp, 0L); - consumer.updateBeginningOffsets(startOffsets); - - // WHEN - countryPopulationConsumer.startBySubscribing(TOPIC); - - // THEN - assertThat(updates).hasSize(1); - assertThat(consumer.closed()).isTrue(); - } - - class CountryPopulation { - - private String country; - private Integer population; - - // standard constructor, getters and setters - - - public CountryPopulation(String country, Integer population) { - this.country = country; - this.population = population; - } - - public String getCountry() { - return country; - } - - public void setCountry(String country) { - this.country = country; - } - - public Integer getPopulation() { - return population; - } - - public void setPopulation(Integer population) { - this.population = population; - } - } - - public class CountryPopulationConsumer { - - private Consumer consumer; - private java.util.function.Consumer exceptionConsumer; - private java.util.function.Consumer countryPopulationConsumer; - - // standard constructor - - - public CountryPopulationConsumer(Consumer consumer, java.util.function.Consumer exceptionConsumer, java.util.function.Consumer countryPopulationConsumer) { - this.consumer = consumer; - this.exceptionConsumer = exceptionConsumer; - this.countryPopulationConsumer = countryPopulationConsumer; - } - - void startBySubscribing(String topic) { - consume(() -> consumer.subscribe(Collections.singleton(topic))); - } - - void startByAssigning(String topic, int partition) { - consume(() -> consumer.assign(Collections.singleton(new TopicPartition(topic, partition)))); - } - - private void consume(Runnable beforePollingTask) { - try { - beforePollingTask.run(); - while (true) { - ConsumerRecords records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS)); - StreamSupport.stream(records.spliterator(), false) - .map(record -> new CountryPopulation(record.key(), record.value())) - .forEach(countryPopulationConsumer); - consumer.commitSync(); - } - } catch (WakeupException e) { - System.out.println("Shutting down..."); - } catch (RuntimeException ex) { - exceptionConsumer.accept(ex); - } finally { - consumer.close(); - } - } - - public void stop() { - consumer.wakeup(); - } - } - - private ConsumerRecord record(String topic, int partition, String country, int population) { - return new ConsumerRecord<>(topic, partition, 0, country, population); - } } \ No newline at end of file