[GitHub] [pinot] KKcorps commented on a change in pull request #8333: Add support for Kinesis datasource in Github Events recipe

[GitBox]

KKcorps commented on a change in pull request #8333:
URL: https://github.com/apache/pinot/pull/8333#discussion_r825699322



##########
File path: 
pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StreamGitHubEventsCommand.java
##########
@@ -19,9 +19,11 @@
 package org.apache.pinot.tools.admin.command;
 
 import org.apache.pinot.spi.plugin.PluginManager;
+import org.apache.pinot.spi.stream.StreamDataProducer;
 import org.apache.pinot.tools.Command;
 import 
org.apache.pinot.tools.streams.githubevents.PullRequestMergedEventsStream;
 import org.apache.pinot.tools.utils.KafkaStarterUtils;
+import org.apache.pinot.tools.utils.StreamSourceType;
 import picocli.CommandLine;
 
 

Review comment:
       done

##########
File path: 
pinot-tools/src/main/java/org/apache/pinot/tools/GitHubEventsQuickstart.java
##########
@@ -40,32 +43,63 @@
 
 /**
  * Sets up a demo Pinot cluster with 1 zookeeper, 1 controller, 1 broker and 1 
server
- * Sets up a demo Kafka cluster, and creates a topic pullRequestMergedEvents
+ * Sets up a demo Kafka/Kinesis cluster, and creates a topic 
pullRequestMergedEvents
  * Creates a realtime table pullRequestMergedEvents
  * Starts the {@link PullRequestMergedEventsStream} to publish 
pullRequestMergedEvents into the topic
  */
 public class GitHubEventsQuickstart extends QuickStartBase {
   private static final Logger LOGGER = 
LoggerFactory.getLogger(GitHubEventsQuickstart.class);
-  private StreamDataServerStartable _kafkaStarter;
+  private StreamDataServerStartable _serverStarter;
   private ZkStarter.ZookeeperInstance _zookeeperInstance;
   private String _personalAccessToken;
+  private StreamSourceType _sourceType;
 
   public GitHubEventsQuickstart() {
   }
 
   private void startKafka() {
     _zookeeperInstance = ZkStarter.startLocalZkServer();
     try {
-      _kafkaStarter = 
StreamDataProvider.getServerDataStartable(KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME,
+      _serverStarter = 
StreamDataProvider.getServerDataStartable(KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME,
           KafkaStarterUtils.getDefaultKafkaConfiguration(_zookeeperInstance));
     } catch (Exception e) {
       throw new RuntimeException("Failed to start " + 
KafkaStarterUtils.KAFKA_SERVER_STARTABLE_CLASS_NAME, e);
     }
-    _kafkaStarter.start();
-    _kafkaStarter.createTopic("pullRequestMergedEvents", 
KafkaStarterUtils.getTopicCreationProps(2));
+    _serverStarter.start();
+    _serverStarter.createTopic("pullRequestMergedEvents", 
KafkaStarterUtils.getTopicCreationProps(2));
   }
 
-  private void execute(String personalAccessToken)
+  private void startKinesis() {
+    try {
+
+      Properties serverProperties = new Properties();
+      serverProperties.put(KinesisStarterUtils.PORT, 4566);
+      _serverStarter =
+          
StreamDataProvider.getServerDataStartable(KinesisStarterUtils.KINESIS_SERVER_STARTABLE_CLASS_NAME,
+              serverProperties);
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to start " + 
KinesisStarterUtils.KINESIS_SERVER_STARTABLE_CLASS_NAME, e);
+    }
+    _serverStarter.start();
+
+    Properties topicProperties = new Properties();
+    topicProperties.put(KinesisStarterUtils.NUM_SHARDS, 3);
+    _serverStarter.createTopic("pullRequestMergedEvents", topicProperties);
+  }
+
+  private void startServer() {

Review comment:
       done




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org