@Test public void testPulsarKafkaProducer() { ClientBuilder mockClientBuilder = mock(ClientBuilder.class); ProducerBuilder mockProducerBuilder = mock(ProducerBuilder.class); doAnswer(invocation -> { Assert.assertEquals((int)invocation.getArguments()[0], 1000000, "Send time out is suppose to be 1000."); return mockProducerBuilder; }).when(mockProducerBuilder).sendTimeout(anyInt(), any(TimeUnit.class)); doReturn(mockClientBuilder).when(mockClientBuilder).serviceUrl(anyString()); doAnswer(invocation -> { Assert.assertEquals((int)invocation.getArguments()[0], 1000, "Keep alive interval is suppose to be 1000."); return mockClientBuilder; }).when(mockClientBuilder).keepAliveInterval(anyInt(), any(TimeUnit.class)); PowerMockito.mockStatic(PulsarClientKafkaConfig.class); PowerMockito.mockStatic(PulsarProducerKafkaConfig.class); when(PulsarClientKafkaConfig.getClientBuilder(any(Properties.class))).thenReturn(mockClientBuilder); when(PulsarProducerKafkaConfig.getProducerBuilder(any(), any())).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"); new PulsarKafkaProducer<>(properties); verify(mockClientBuilder, times(1)).keepAliveInterval(1000, TimeUnit.SECONDS); verify(mockProducerBuilder, times(1)).sendTimeout(1000000, TimeUnit.MILLISECONDS); }
public PulsarKafkaProducer(Map<String, Object> configs) { this(new ProducerConfig(configs), null, null); }
PulsarKafkaProducer implements Producer<K, V> { public PulsarKafkaProducer(Map<String, Object> configs) { this(new ProducerConfig(configs), null, null); } }
PulsarKafkaProducer implements Producer<K, V> { public PulsarKafkaProducer(Map<String, Object> configs) { this(new ProducerConfig(configs), null, 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> { public PulsarKafkaProducer(Map<String, Object> configs) { this(new ProducerConfig(configs), null, 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> { public PulsarKafkaProducer(Map<String, Object> configs) { this(new ProducerConfig(configs), null, 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);  }