@Test public void send() { byte[] payload = { 1, 2, 3 }; when(kafkaMessage.getPayload()).thenReturn(payload); when(kafkaMessage.getQualifiedTableName()).thenReturn("database.table"); when(producer.partitionsFor("topic")).thenReturn(partitionInfoList); when(partitionInfoList.size()).thenReturn(5); KafkaMessageSender kafkaMessageSender = new KafkaMessageSender("topic", producer); kafkaMessageSender.send(kafkaMessage); verify(producer).send(producerRecordCaptor.capture()); ProducerRecord record = producerRecordCaptor.getValue(); assertThat(record.topic()).isEqualToIgnoringCase("topic"); assertThat(record.partition()).isEqualTo(1); assertThat(record.value()).isEqualTo(payload); }
public void send(KafkaMessage kafkaMessage) { int partition = Math.abs(kafkaMessage.getQualifiedTableName().hashCode() % numberOfPartitions); producer.send(new ProducerRecord<>(topic, partition, kafkaMessage.getTimestamp(), kafkaMessage.getPayload())); }
KafkaMessageSender { public void send(KafkaMessage kafkaMessage) { int partition = Math.abs(kafkaMessage.getQualifiedTableName().hashCode() % numberOfPartitions); producer.send(new ProducerRecord<>(topic, partition, kafkaMessage.getTimestamp(), kafkaMessage.getPayload())); } }
KafkaMessageSender { public void send(KafkaMessage kafkaMessage) { int partition = Math.abs(kafkaMessage.getQualifiedTableName().hashCode() % numberOfPartitions); producer.send(new ProducerRecord<>(topic, partition, kafkaMessage.getTimestamp(), kafkaMessage.getPayload())); } KafkaMessageSender(Configuration conf); @VisibleForTesting  KafkaMessageSender(String topic, KafkaProducer<Long, byte[]> producer); }
KafkaMessageSender { public void send(KafkaMessage kafkaMessage) { int partition = Math.abs(kafkaMessage.getQualifiedTableName().hashCode() % numberOfPartitions); producer.send(new ProducerRecord<>(topic, partition, kafkaMessage.getTimestamp(), kafkaMessage.getPayload())); } KafkaMessageSender(Configuration conf); @VisibleForTesting  KafkaMessageSender(String topic, KafkaProducer<Long, byte[]> producer); void send(KafkaMessage kafkaMessage); }
KafkaMessageSender { public void send(KafkaMessage kafkaMessage) { int partition = Math.abs(kafkaMessage.getQualifiedTableName().hashCode() % numberOfPartitions); producer.send(new ProducerRecord<>(topic, partition, kafkaMessage.getTimestamp(), kafkaMessage.getPayload())); } KafkaMessageSender(Configuration conf); @VisibleForTesting  KafkaMessageSender(String topic, KafkaProducer<Long, byte[]> producer); void send(KafkaMessage kafkaMessage);  }