Copilot commented on code in PR #18687:
URL: https://github.com/apache/pinot/pull/18687#discussion_r3360226985
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerTest.java:
##########
@@ -286,6 +287,39 @@ public void testSubsetPartitionConsumption() throws
TimeoutException {
}
}
+ @Test
+ public void testComputePartitionGroupMetadataUsesKafkaPartitionIds()
+ throws Exception {
+ String streamType = "kafka";
+ String streamKafkaBrokerList = _kafkaBrokerAddress;
+ String clientId = "partition-id-client";
+ String tableNameWithType = "tableName_REALTIME";
+
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("streamType", streamType);
+ streamConfigMap.put("stream.kafka.topic.name",
TEST_TOPIC_SUBSET_PARTITION);
+ streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+ streamConfigMap.put("stream.kafka.consumer.factory.class.name",
getKafkaConsumerFactoryName());
+ streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+ StreamConfig streamConfig = new StreamConfig(tableNameWithType,
streamConfigMap);
+
+ List<PartitionGroupConsumptionStatus> currentStatuses = List.of(
+ new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(1, 1, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(3, 3, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(4, 4, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(5, 5, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(6, 6, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(7, 7, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"));
+
+ KafkaStreamMetadataProvider streamMetadataProvider = new
KafkaStreamMetadataProvider(clientId, streamConfig);
+ List<PartitionGroupMetadata> partitionGroupMetadataList =
+ streamMetadataProvider.computePartitionGroupMetadata(clientId,
streamConfig, currentStatuses, 10000);
+
+
assertEquals(partitionGroupMetadataList.stream().map(PartitionGroupMetadata::getPartitionGroupId)
+ .collect(Collectors.toList()), List.of(0, 1, 2, 3, 4, 5, 6, 7));
+ }
Review Comment:
`KafkaStreamMetadataProvider` holds a Kafka consumer (via
`KafkaPartitionLevelConnectionHandler.close()`), but this new test instantiates
it without closing. That can leak threads/connections and make the test suite
flaky; wrap it in try-with-resources.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java:
##########
@@ -282,6 +283,39 @@ public void testSubsetPartitionConsumption()
}
}
+ @Test
+ public void testComputePartitionGroupMetadataUsesKafkaPartitionIds()
+ throws Exception {
+ String streamType = "kafka";
+ String streamKafkaBrokerList = _kafkaBrokerAddress;
+ String clientId = "partition-id-client";
+ String tableNameWithType = "tableName_REALTIME";
+
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("streamType", streamType);
+ streamConfigMap.put("stream.kafka.topic.name",
TEST_TOPIC_SUBSET_PARTITION);
+ streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+ streamConfigMap.put("stream.kafka.consumer.factory.class.name",
getKafkaConsumerFactoryName());
+ streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+ StreamConfig streamConfig = new StreamConfig(tableNameWithType,
streamConfigMap);
+
+ List<PartitionGroupConsumptionStatus> currentStatuses = List.of(
+ new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(1, 1, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(3, 3, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(4, 4, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(5, 5, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(6, 6, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"),
+ new PartitionGroupConsumptionStatus(7, 7, new LongMsgOffset(0), new
LongMsgOffset(10), "DONE"));
+
+ KafkaStreamMetadataProvider streamMetadataProvider = new
KafkaStreamMetadataProvider(clientId, streamConfig);
+ List<PartitionGroupMetadata> partitionGroupMetadataList =
+ streamMetadataProvider.computePartitionGroupMetadata(clientId,
streamConfig, currentStatuses, 10000);
+
+
assertEquals(partitionGroupMetadataList.stream().map(PartitionGroupMetadata::getPartitionGroupId)
+ .collect(Collectors.toList()), List.of(0, 1, 2, 3, 4, 5, 6, 7));
+ }
Review Comment:
`KafkaStreamMetadataProvider` implements `Closeable` (inherited `close()`
shuts down the underlying Kafka consumer). This new test creates an instance
and uses it without closing, which can leak resources and lead to flaky tests;
use try-with-resources here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
