abossert commented on a change in pull request #8112: removed hard-coded Kafka
key and value deserializer
URL: https://github.com/apache/incubator-druid/pull/8112#discussion_r308918954
##########
File path:
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaRecordSupplierTest.java
##########
@@ -183,6 +205,37 @@ public void testSupplierSetup() throws
ExecutionException, InterruptedException
recordSupplier.close();
}
+
+ @Test
+ public void testSupplierSetupCustomDeserializer() throws ExecutionException,
InterruptedException
+ {
+
+ // Insert data
+ insertData();
+
+ Set<StreamPartition<Integer>> partitions = ImmutableSet.of(
+ StreamPartition.of(topic, 0),
+ StreamPartition.of(topic, 1)
+ );
+
+ Map<String, Object> properties = kafkaServer.consumerProperties();
+ properties.put("key.deserializer",
KafkaRecordSupplierTest.TestKafkaDeserializer.class.getName());
+ properties.put("value.deserializer",
KafkaRecordSupplierTest.TestKafkaDeserializer.class.getName());
+
+ KafkaRecordSupplier recordSupplier = new KafkaRecordSupplier(
+ properties,
+ objectMapper
+ );
+
+ Assert.assertTrue(recordSupplier.getAssignment().isEmpty());
+
+ recordSupplier.assign(partitions);
+
+ Assert.assertEquals(partitions, recordSupplier.getAssignment());
Review comment:
Aha! I misunderstood what the recordsupplier test was doing under the hood.
I have addressed this with an additional test...please let me know if that now
covers the needed functionality.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
