@Test public void testPulsarKafkaSendAvro() throws PulsarClientException { PulsarClient mockClient = mock(PulsarClient.class); ProducerBuilder mockProducerBuilder = mock(ProducerBuilder.class); org.apache.pulsar.client.api.Producer mockProducer = mock(org.apache.pulsar.client.api.Producer.class); ClientBuilder mockClientBuilder = mock(ClientBuilder.class); CompletableFuture mockPartitionFuture = new CompletableFuture(); CompletableFuture mockSendAsyncFuture = new CompletableFuture(); TypedMessageBuilder mockTypedMessageBuilder = mock(TypedMessageBuilderImpl.class); mockPartitionFuture.complete(new ArrayList<>()); mockSendAsyncFuture.complete(new MessageIdImpl(1, 1, 1)); doReturn(mockClientBuilder).when(mockClientBuilder).serviceUrl(anyString()); doReturn(mockClientBuilder).when(mockClientBuilder).keepAliveInterval(anyInt(), any(TimeUnit.class)); doReturn(mockClient).when(mockClientBuilder).build(); doReturn(mockPartitionFuture).when(mockClient).getPartitionsForTopic(anyString()); doReturn(mockProducerBuilder).when(mockProducerBuilder).topic(anyString()); doReturn(mockProducerBuilder).when(mockProducerBuilder).clone(); doReturn(mockProducer).when(mockProducerBuilder).create(); doReturn(mockTypedMessageBuilder).when(mockProducer).newMessage(); doReturn(mockSendAsyncFuture).when(mockTypedMessageBuilder).sendAsync(); PowerMockito.mockStatic(PulsarClientKafkaConfig.class); PowerMockito.mockStatic(PulsarProducerKafkaConfig.class); when(PulsarClientKafkaConfig.getClientBuilder(any(Properties.class))).thenReturn(mockClientBuilder); when(PulsarProducerKafkaConfig.getProducerBuilder(any(PulsarClient.class), any(Properties.class))).thenReturn(mockProducerBuilder); Properties properties = new Properties(); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Arrays.asList("pulsar: properties.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, "1000000"); properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "1000000"); AvroSchema<Bar> barSchema = AvroSchema.of(SchemaDefinition.<Bar>builder().withPojo(Bar.class).build()); AvroSchema<Foo> fooSchema = AvroSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build()); PulsarKafkaProducer<Foo, Bar> pulsarKafkaProducer = new PulsarKafkaProducer<>(properties, fooSchema, barSchema); Bar bar = new Bar(); bar.setField1(true); Foo foo = new Foo(); foo.setField1("field1"); foo.setField2("field2"); foo.setField3(3); pulsarKafkaProducer.send(new ProducerRecord<>("topic", 1, foo, bar)); verify(mockTypedMessageBuilder).sendAsync(); }
@Override public Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord) { return send(producerRecord, null); }
PulsarKafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord) { return send(producerRecord, null); } }
PulsarKafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord) { return send(producerRecord, null); } PulsarKafkaProducer(Map<String, Object> configs);  PulsarKafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer,
                               Serializer<V> valueSerializer);  PulsarKafkaProducer(Map<String, Object> configs, Schema<K> keySchema, Schema<V> valueSchema);  PulsarKafkaProducer(Properties properties);  PulsarKafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer);  PulsarKafkaProducer(Properties properties, Schema<K> keySchema, Schema<V> valueSchema); @SuppressWarnings({ "unchecked", "deprecation" }) private  PulsarKafkaProducer(ProducerConfig producerConfig, Schema<K> keySchema, Schema<V> valueSchema); }
PulsarKafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord) { return send(producerRecord, null); } PulsarKafkaProducer(Map<String, Object> configs);  PulsarKafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer,
                               Serializer<V> valueSerializer);  PulsarKafkaProducer(Map<String, Object> configs, Schema<K> keySchema, Schema<V> valueSchema);  PulsarKafkaProducer(Properties properties);  PulsarKafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer);  PulsarKafkaProducer(Properties properties, Schema<K> keySchema, Schema<V> valueSchema); @SuppressWarnings({ "unchecked", "deprecation" }) private  PulsarKafkaProducer(ProducerConfig producerConfig, Schema<K> keySchema, Schema<V> valueSchema); @Override Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord); @Override Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord, Callback callback); @Override void flush(); @Override List<PartitionInfo> partitionsFor(String s); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); @Override void close(long timeout, TimeUnit unit); }
PulsarKafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord) { return send(producerRecord, null); } PulsarKafkaProducer(Map<String, Object> configs);  PulsarKafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer,
                               Serializer<V> valueSerializer);  PulsarKafkaProducer(Map<String, Object> configs, Schema<K> keySchema, Schema<V> valueSchema);  PulsarKafkaProducer(Properties properties);  PulsarKafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer);  PulsarKafkaProducer(Properties properties, Schema<K> keySchema, Schema<V> valueSchema); @SuppressWarnings({ "unchecked", "deprecation" }) private  PulsarKafkaProducer(ProducerConfig producerConfig, Schema<K> keySchema, Schema<V> valueSchema); @Override Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord); @Override Future<RecordMetadata> send(ProducerRecord<K, V> producerRecord, Callback callback); @Override void flush(); @Override List<PartitionInfo> partitionsFor(String s); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); @Override void close(long timeout, TimeUnit unit);  }