@Test @SuppressWarnings("rawtypes") public void testFlakyConsumer() throws Exception { String topic = "Pizza2"; createTopic(_zkUtils, topic); LOG.info("Sending first event, to avoid an empty topic."); produceEvents(_kafkaCluster, _zkUtils, topic, 0, 1); LOG.info("Creating and Starting KafkaConnectorTask"); Datastream datastream = getDatastream(_broker, topic); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); task.setZkAdapter(Mockito.mock(ZkAdapter.class)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); task.setEventProducer(datastreamProducer); KafkaConnectorTask connectorTask = createKafkaConnectorTask(task); KafkaConnectorTask spiedConnectorTask = Mockito.spy(connectorTask); KafkaConsumer mockKafkaConsumer = Mockito.mock(KafkaConsumer.class); doReturn(mockKafkaConsumer).when(spiedConnectorTask).createKafkaConsumer(Mockito.any()); boolean exceptionThrown = false; try { spiedConnectorTask.run(); } catch (DatastreamRuntimeException ex) { exceptionThrown = true; } Assert.assertTrue(exceptionThrown); verify(spiedConnectorTask, Mockito.atLeast(5)).handlePollRecordsException(any()); }
@Override protected Consumer<?, ?> createKafkaConsumer(Properties consumerProps) { return createConsumer(_consumerFactory, consumerProps, getKafkaGroupId(_datastreamTask, _groupIdConstructor, _consumerMetrics, LOG), _srcConnString); }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @Override protected Consumer<?, ?> createKafkaConsumer(Properties consumerProps) { return createConsumer(_consumerFactory, consumerProps, getKafkaGroupId(_datastreamTask, _groupIdConstructor, _consumerMetrics, LOG), _srcConnString); } }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @Override protected Consumer<?, ?> createKafkaConsumer(Properties consumerProps) { return createConsumer(_consumerFactory, consumerProps, getKafkaGroupId(_datastreamTask, _groupIdConstructor, _consumerMetrics, LOG), _srcConnString); } KafkaConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      GroupIdConstructor groupIdConstructor); }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @Override protected Consumer<?, ?> createKafkaConsumer(Properties consumerProps) { return createConsumer(_consumerFactory, consumerProps, getKafkaGroupId(_datastreamTask, _groupIdConstructor, _consumerMetrics, LOG), _srcConnString); } KafkaConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      GroupIdConstructor groupIdConstructor); static Consumer<?, ?> createConsumer(KafkaConsumerFactory<?, ?> consumerFactory, Properties consumerProps,
      String groupId, KafkaConnectionString connectionString); static List<BrooklinMetricInfo> getMetricInfos(String connectorName); }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @Override protected Consumer<?, ?> createKafkaConsumer(Properties consumerProps) { return createConsumer(_consumerFactory, consumerProps, getKafkaGroupId(_datastreamTask, _groupIdConstructor, _consumerMetrics, LOG), _srcConnString); } KafkaConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      GroupIdConstructor groupIdConstructor); static Consumer<?, ?> createConsumer(KafkaConsumerFactory<?, ?> consumerFactory, Properties consumerProps,
      String groupId, KafkaConnectionString connectionString); static List<BrooklinMetricInfo> getMetricInfos(String connectorName);  }