我在 JUnit 測試中啟動了一個 EmbeddedKafka 實例。我可以在應用程序中正確讀取已推送到流的記錄，但我注意到的一件事是每個主題只有一個分區。誰能解釋為什么？在我的應用程序中，我有以下內容：List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);這將返回一個包含一項的列表。當針對具有 3 個分區的本地 Kafka 運行時，它會按預期返回包含 3 個項目的列表。我的測試看起來像：@RunWith(SpringRunner.class)@SpringBootTest@EmbeddedKafka(partitions = 3)@ActiveProfiles("inmemory")@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)@TestPropertySource( locations = "classpath:application-test.properties", properties = {"app.onlyMonitorIfDataUpdated=true"})public class MonitorRestKafkaIntegrationTest { @Autowired private EmbeddedKafkaBroker embeddedKafkaBroker; @Value("${spring.embedded.kafka.brokers}") private String embeddedBrokers; @Autowired private WebApplicationContext wac; @Autowired private JsonUtility jsonUtility; private MockMvc mockMvc; @Before public void setup() { mockMvc = webAppContextSetup(wac).build(); UserGroupInformation.setLoginUser(UserGroupInformation.createRemoteUser("dummyUser")); } private ResultActions interactiveMonitoringREST(String eggID, String monitoringParams) throws Exception { return mockMvc.perform(post(String.format("/eggs/%s/interactive", eggID)).contentType(MediaType.APPLICATION_JSON_VALUE).content(monitoringParams)); }
2 回答
慕容3067478
TA貢獻1773條經驗 獲得超3個贊
您需要告訴經紀人預先創建主題...
@SpringBootTest
@EmbeddedKafka(topics = "foo", partitions = 3)
class So57481979ApplicationTests {
@Test
void testPartitions(@Autowired KafkaAdmin admin) throws InterruptedException, ExecutionException {
AdminClient client = AdminClient.create(admin.getConfig());
Map<String, TopicDescription> map = client.describeTopics(Collections.singletonList("foo")).all().get();
System.out.println(map.values().iterator().next().partitions().size());
}
}
num.partitions或者，如果您希望代理在首次使用時為您自動創建主題，則設置代理屬性。
我們可能應該根據分區屬性自動執行此操作。
慕絲7291255
TA貢獻1859條經驗 獲得超6個贊
我發現bootstrapServersPropertyis important in @EmbeddedKafka，它用于填充 中的屬性application-test.yml，然后可用于創建消費者/偵聽器容器。
添加回答
舉報
0/150
提交
取消