### Question:
KafkaEventReceiver implements EventReceiver { void consume(EventReceiverListener listener) { var currentOffset = new AtomicLong(0L); var progressLogger = executorService .scheduleAtFixedRate(() -> log.info("Current offset {}", currentOffset.get()), 10, 10, SECONDS); var topicPartition = new TopicPartition(config.getTopic(), 0); var topicPartitions = List.of(topicPartition); int partitions = consumer.partitionsFor(topicPartition.topic()).size(); if (partitions != 1) { throw new IllegalStateException("Unsupported partition count. Require 1, got " + partitions); } long beginningOffset = consumer.beginningOffsets(topicPartitions).get(topicPartition); long endOffset = consumer.endOffsets(topicPartitions).get(topicPartition); log.info("Offsets: beginning[{}], end[{}]", beginningOffset, endOffset); consumer.assign(topicPartitions); consumer.seekToBeginning(topicPartitions); var loaded = false; if (endOffset == 0L) { progressLogger.cancel(true); log.info("Loading complete. Empty topic."); listener.onEvent(LOAD_COMPLETE); loaded = true; } while (!shuttingDown) { for (ConsumerRecord<AvroKey, AvroValue> record : consumer.poll(Duration.ofMillis(100))) { var event = converter.toModel(record.key(), record.value()); currentOffset.set(record.offset()); try { listener.onEvent(event); } catch (Exception e) { log.error("Listener failed for event {}", event, e); } receiveCorrelationId(record); if (!loaded && record.offset() >= endOffset - 1L) { progressLogger.cancel(true); log.info("Loading complete. Reached offset " + record.offset()); listener.onEvent(LOAD_COMPLETE); loaded = true; } } } } KafkaEventReceiver(Config config, EventCorrelator correlator);  KafkaEventReceiver(Config config); @Override void receive(EventReceiverListener listener); @Override void close();  }### Answer:
@Test(expected = IllegalStateException.class) public void incorrectNumberOfPartitions() { when(consumer.partitionsFor(topic)).thenReturn(List.of()); underTest.consume(listener); }