Реализация пользовательского потребителя Kafka на основе LinkedBlockingQueue не может вернуть размер ⇐ JAVA

[Anonymous]

обнаружил, что kafkaconsumer довольно неудобен для тестирования, пытался реализовать его сам на основе LinkedBlockqueue.

Код: Выделить всё

@Slf4j
@Component
public class KafkaConsumerCashOut {
private final BlockingQueue queue = new LinkedBlockingQueue();

private final KafkaConsumer consumer;
private final String topic;

private final AtomicBoolean isShutdown = new AtomicBoolean(false);
private final ExecutorService poller = Executors.newSingleThreadExecutor();

public KafkaConsumerCashOut(TestKafkaConsumerProperties kafkaStreamProperties, CashOutTopicProperties cashOutTopicProperties) {
this.topic = cashOutTopicProperties.getTopic();
this.consumer = new KafkaConsumer(new HashMap(kafkaStreamProperties.getProperties()));
}

@SneakyThrows
public KeyValue take() {
return queue.take();
}

@SneakyThrows
public @Nullable KeyValue poll(Duration timeout) {
return queue.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
}

public int size() {
return queue.size();
}

@PreDestroy
public void destroy() throws InterruptedException {
log.info("Closing consumer ...");
isShutdown.compareAndSet(false, true);
poller.awaitTermination(5, TimeUnit.SECONDS);
}

@PostConstruct
void init() {
poller.submit(() -> {
try {
consumer.subscribe(Collections.singleton(topic));
while (!isShutdown.get()) {
var polled = consumer.poll(Duration.ofMillis(100));
if (!polled.isEmpty()) {
log.info("Kafka consumer is polled records {}", polled.count());
for (var record : polled) {
Preconditions.checkState(queue.offer(KeyValue.of(record.key(), record.value())));
}
}
}
} catch (Throwable th) {
log.warn("Error in consumer", th);
} finally {
log.info("Consumer loop terminated");
try {
consumer.close();
} catch (Throwable th) {
log.warn("Error closing consumer", th);
}
}
});
}
}

Я использую два разных свойства: одно для получения конфигурации потребителя, а другое для получения имени темы.

Код: Выделить всё

@Data
@RequiredArgsConstructor(staticName = "of")
public class KeyValue {
private final K key;
private final V message;
}
< /code>
И основная функциональность работает нормально.

wehper, вы можете увидеть фрагмент из теста: < /p>
@Autowired
KafkaConsumer watchResult;

//...
var result = watchResult.take();
assertThat(result.getKey()).isEqualTo(OUTCOME_ID_STRING);
assertThat(result.getMessage()).isEqualTo(expectedResult);
< /code>
И он работает нормально. Утверждения прошли нормально. Даже если вы добавите некоторую информацию о журнале, например: < /p>
@SneakyThrows
public KeyValue take() {
KeyValue element = queue.take();
log.debug("take element: {}", element);
return element;
}
< /code>
Он будет напечатан на консоли: < /p>
take element: KeyValue(key=241389702681395312, message={"outcomeId": 241389702681395312, "drawResult": true})

Один пункт, который я не мог сделать, - это просто добавление еще одного утверждения, прежде чем взять элемент с помощью take () :
assertThat(watchResultCashOut.size()).isEqualTo(1);
< /code>
Хотя элемент действительно получен нормально, это утверждение не удается. Невозможно найти то, чего не хватает.
Как решить эту проблему?

Подробнее здесь: https://stackoverflow.com/questions/797 ... t-return-s