[
https://issues.apache.org/jira/browse/TAJO-1480?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15424350#comment-15424350
]
ASF GitHub Bot commented on TAJO-1480:
--------------------------------------
Github user jinossy commented on a diff in the pull request:
https://github.com/apache/tajo/pull/1041#discussion_r75107466
--- Diff:
tajo-storage/tajo-storage-kafka/src/test/java/org/apache/tajo/storage/kafka/server/EmbeddedZookeeper.java
---
@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage.kafka.server;
+
+import static io.airlift.testing.FileUtils.deleteRecursively;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+
+import com.google.common.io.Files;
+
+public class EmbeddedZookeeper implements Closeable {
+ private final int port;
+ private final File zkDataDir;
+ private final ZooKeeperServer zkServer;
+ private final NIOServerCnxnFactory cnxnFactory;
+
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final AtomicBoolean stopped = new AtomicBoolean();
+
+ public EmbeddedZookeeper() throws IOException {
+ this(2181);
+ }
+
+ public EmbeddedZookeeper(int port) throws IOException {
+ this.port = port;
+ zkDataDir = Files.createTempDir();
+ zkServer = new ZooKeeperServer();
+
+ FileTxnSnapLog ftxn = new FileTxnSnapLog(zkDataDir, zkDataDir);
+ zkServer.setTxnLogFactory(ftxn);
+
+ cnxnFactory = new NIOServerCnxnFactory();
+ cnxnFactory.configure(new InetSocketAddress(port), 0);
+ }
+
+ public void start() throws InterruptedException, IOException {
+ if (!started.getAndSet(true)) {
+ cnxnFactory.startup(zkServer);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (started.get() && !stopped.getAndSet(true)) {
+ cnxnFactory.shutdown();
+ try {
+ cnxnFactory.join();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ if (zkServer.isRunning()) {
+ zkServer.shutdown();
+ }
+ deleteRecursively(zkDataDir);
+ }
+ }
+
+ public String getConnectString() {
+ return "127.0.0.1:" + Integer.toString(port);
--- End diff --
Remove hard coded address.
> Kafka Consumer for kafka strage.
> --------------------------------
>
> Key: TAJO-1480
> URL: https://issues.apache.org/jira/browse/TAJO-1480
> Project: Tajo
> Issue Type: Sub-task
> Components: Storage
> Reporter: YeonSu Han
> Assignee: Byunghwa Yun
> Labels: kafka_storage
>
> Scanner of kafka storage needs to implement a Kafka Consumer for to fetch
> data from kafka.
> The Kafka Consumer have methods like this,
> - getPartitions(): Get partition id list from specific topic.
> - poll(): Fetch data from kafka.
> - getEarliestOffset(): Get earliest offset from specific partition.
> - getLatestOffset(): Get latest offset from specific partition.
> - etc..
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
