Allow realtime data providers to accept non-kafka producers

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java

@@ -190,7 +190,7 @@ public void setUp()
       // creation failure if both calls entered ProcessNewSegment. In/such cases ensure that we upload all the
       // segments again/to ensure that the data is setup correctly.
       assertTrue(e.getMessage().contains("Another segment upload is in progress for segment") || e.getMessage()
-          .contains("Failed to create ZK metadata for segment"));
+          .contains("Failed to create ZK metadata for segment"), e.getMessage());
       uploadSegments(getTableName(), _tarDir);
     }
 

pinot-tools/src/main/java/org/apache/pinot/tools/GitHubEventsQuickstart.java

@@ -104,7 +104,7 @@ private void execute(String personalAccessToken)
     printStatus(Color.CYAN, "***** Starting pullRequestMergedEvents data stream and publishing to Kafka *****");
     final PullRequestMergedEventsStream pullRequestMergedEventsStream =
         new PullRequestMergedEventsStream(schemaFile.getAbsolutePath(), "pullRequestMergedEvents",
-            KafkaStarterUtils.DEFAULT_KAFKA_BROKER, personalAccessToken);
+            personalAccessToken, PullRequestMergedEventsStream.getKafkaStreamDataProducer());
     pullRequestMergedEventsStream.execute();
     printStatus(Color.CYAN, "***** Waiting for 10 seconds for a few events to get populated *****");
     Thread.sleep(10000);

pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StreamGitHubEventsCommand.java

@@ -105,7 +105,8 @@ public boolean execute()
     PluginManager.get().init();
     if (PULL_REQUEST_MERGED_EVENT_TYPE.equals(_eventType)) {
       PullRequestMergedEventsStream pullRequestMergedEventsStream =
-          new PullRequestMergedEventsStream(_schemaFile, _topic, _kafkaBrokerList, _personalAccessToken);
+          new PullRequestMergedEventsStream(_schemaFile, _topic, _personalAccessToken,
+              PullRequestMergedEventsStream.getKafkaStreamDataProducer(_kafkaBrokerList));
       pullRequestMergedEventsStream.execute();
     } else {
       throw new UnsupportedOperationException("Event type " + _eventType + " is unsupported");

pinot-tools/src/main/java/org/apache/pinot/tools/streams/AirlineDataStream.java

@@ -59,21 +59,31 @@ public class AirlineDataStream {
 
   public AirlineDataStream(Schema pinotSchema, TableConfig tableConfig, File avroFile)
       throws Exception {
+    this(pinotSchema, tableConfig, avroFile, getDefaultKafkaProducer());
+  }
+
+  public AirlineDataStream(Schema pinotSchema, TableConfig tableConfig, File avroFile, StreamDataProducer producer)
+      throws IOException {
     _pinotSchema = pinotSchema;
     _timeColumnName = tableConfig.getValidationConfig().getTimeColumnName();
     _avroFile = avroFile;
     createStream();
+    _producer = producer;
+    _service = Executors.newFixedThreadPool(1);
+    QuickStartBase.printStatus(Quickstart.Color.YELLOW,
+        "***** Offine data has max time as 16101, realtime will start consuming from time 16102 and increment time "
+            + "every 60 events (which is approximately 60 seconds) *****");
+  }
+
+  public static StreamDataProducer getDefaultKafkaProducer()
+      throws Exception {
     Properties properties = new Properties();
     properties.put("metadata.broker.list", KafkaStarterUtils.DEFAULT_KAFKA_BROKER);
     properties.put("serializer.class", "kafka.serializer.DefaultEncoder");
     properties.put("request.required.acks", "1");
 
-    _producer = StreamDataProvider.getStreamDataProducer(KafkaStarterUtils.KAFKA_PRODUCER_CLASS_NAME, properties);
+    return StreamDataProvider.getStreamDataProducer(KafkaStarterUtils.KAFKA_PRODUCER_CLASS_NAME, properties);
 
-    _service = Executors.newFixedThreadPool(1);
-    QuickStartBase.printStatus(Quickstart.Color.YELLOW,
-        "***** Offine data has max time as 16101, realtime will start consuming from time 16102 and increment time "
-            + "every 60 events (which is approximately 60 seconds) *****");
   }
 
   public void shutdown() {

pinot-tools/src/main/java/org/apache/pinot/tools/streams/githubevents/PullRequestMergedEventsStream.java

@@ -61,10 +61,9 @@ public class PullRequestMergedEventsStream {
 
   private StreamDataProducer _producer;
 
-  public PullRequestMergedEventsStream(String schemaFilePath, String topicName, String kafkaBrokerList,
-      String personalAccessToken)
+  public PullRequestMergedEventsStream(String schemaFilePath, String topicName, String personalAccessToken,
+      StreamDataProducer producer)
       throws Exception {
-
     _service = Executors.newFixedThreadPool(2);
     try {
       File pinotSchema;
@@ -83,12 +82,21 @@ public PullRequestMergedEventsStream(String schemaFilePath, String topicName, St
     }
     _topicName = topicName;
     _gitHubAPICaller = new GitHubAPICaller(personalAccessToken);
+    _producer = producer;
+  }
+
+  public static StreamDataProducer getKafkaStreamDataProducer()
+      throws Exception {
+    return getKafkaStreamDataProducer(KafkaStarterUtils.DEFAULT_KAFKA_BROKER);
+  }
 
+  public static StreamDataProducer getKafkaStreamDataProducer(String kafkaBrokerList)
+      throws Exception {
     Properties properties = new Properties();
     properties.put("metadata.broker.list", kafkaBrokerList);
     properties.put("serializer.class", "kafka.serializer.DefaultEncoder");
     properties.put("request.required.acks", "1");
-    _producer = StreamDataProvider.getStreamDataProducer(KafkaStarterUtils.KAFKA_PRODUCER_CLASS_NAME, properties);
+    return StreamDataProvider.getStreamDataProducer(KafkaStarterUtils.KAFKA_PRODUCER_CLASS_NAME, properties);
   }
 
   public static void main(String[] args)
@@ -97,8 +105,7 @@ public static void main(String[] args)
     String schemaFile = args[1];
     String topic = "pullRequestMergedEvent";
     PullRequestMergedEventsStream stream =
-        new PullRequestMergedEventsStream(schemaFile, topic, KafkaStarterUtils.DEFAULT_KAFKA_BROKER,
-            personalAccessToken);
+        new PullRequestMergedEventsStream(schemaFile, topic, personalAccessToken, getKafkaStreamDataProducer());
     stream.execute();
   }