@Test public void testAssign() { String topicName = getUniqueTopicName(); KafkaTransportProviderAdmin provider = new KafkaTransportProviderAdmin("test", _transportProviderProperties); int numberOfPartitions = 32; Datastream ds = DatastreamTestUtils.createDatastream("test", "ds1", "source", provider.getDestination(null, topicName), numberOfPartitions); Set<KafkaProducerWrapper<byte[], byte[]>> producers = IntStream.range(0, KafkaTransportProviderAdmin.DEFAULT_PRODUCERS_PER_CONNECTOR) .mapToObj(x -> ((KafkaTransportProvider) provider.assignTransportProvider( new DatastreamTaskImpl(Collections.singletonList(ds)))).getProducers().get(0)) .collect(Collectors.toSet()); Assert.assertEquals(producers.size(), KafkaTransportProviderAdmin.DEFAULT_PRODUCERS_PER_CONNECTOR); Set<KafkaProducerWrapper<byte[], byte[]>> producers2 = IntStream.range(0, KafkaTransportProviderAdmin.DEFAULT_PRODUCERS_PER_CONNECTOR) .mapToObj(x -> ((KafkaTransportProvider) provider.assignTransportProvider( new DatastreamTaskImpl(Collections.singletonList(ds)))).getProducers().get(0)) .collect(Collectors.toSet()); Assert.assertTrue(producers2.containsAll(producers)); producers.forEach(x -> Assert.assertEquals(x.getTasksSize(), 2)); }
public List<KafkaProducerWrapper<byte[], byte[]>> getProducers() { return _producers; }
KafkaTransportProvider implements TransportProvider { public List<KafkaProducerWrapper<byte[], byte[]>> getProducers() { return _producers; } }
KafkaTransportProvider implements TransportProvider { public List<KafkaProducerWrapper<byte[], byte[]>> getProducers() { return _producers; } KafkaTransportProvider(DatastreamTask datastreamTask, List<KafkaProducerWrapper<byte[], byte[]>> producers,
      Properties props, String metricsNamesPrefix); }
KafkaTransportProvider implements TransportProvider { public List<KafkaProducerWrapper<byte[], byte[]>> getProducers() { return _producers; } KafkaTransportProvider(DatastreamTask datastreamTask, List<KafkaProducerWrapper<byte[], byte[]>> producers,
      Properties props, String metricsNamesPrefix); List<KafkaProducerWrapper<byte[], byte[]>> getProducers(); @Override void send(String destinationUri, DatastreamProducerRecord record, SendCallback onSendComplete); @Override void close(); @Override void flush(); static List<BrooklinMetricInfo> getMetricInfos(String metricsNamesPrefix); }
KafkaTransportProvider implements TransportProvider { public List<KafkaProducerWrapper<byte[], byte[]>> getProducers() { return _producers; } KafkaTransportProvider(DatastreamTask datastreamTask, List<KafkaProducerWrapper<byte[], byte[]>> producers,
      Properties props, String metricsNamesPrefix); List<KafkaProducerWrapper<byte[], byte[]>> getProducers(); @Override void send(String destinationUri, DatastreamProducerRecord record, SendCallback onSendComplete); @Override void close(); @Override void flush(); static List<BrooklinMetricInfo> getMetricInfos(String metricsNamesPrefix);  }