本文整理汇总了Java中kafka.utils.TestZKUtils类的典型用法代码示例。如果您正苦于以下问题：Java TestZKUtils类的具体用法？Java TestZKUtils怎么用？Java TestZKUtils使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
TestZKUtils类属于kafka.utils包，在下文中一共展示了TestZKUtils类的9个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。
示例1: initialize
import kafka.utils.TestZKUtils; //导入依赖的package包/类
public void initialize() {
if (initialized) {
throw new IllegalStateException("Context has been already initialized");
}
zkServer = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect());
zkClient = new ZkClient(zkServer.connectString(), 10000, 10000, ZKStringSerializer$.MODULE$);
port = TestUtils.choosePort();
KafkaConfig config = new KafkaConfig(TestUtils.createBrokerConfig(brokerId, port, true));
Time mock = new MockTime();
kafkaServer = new KafkaServer(config, mock);
kafkaServer.startup();
initialized = true;
}
开发者ID:researchgate，项目名称:kafka-metamorph，代码行数:18，代码来源:Kafka08TestContext.java
示例2: start
import kafka.utils.TestZKUtils; //导入依赖的package包/类
public void start() throws IOException {
String zkConnect = TestZKUtils.zookeeperConnect();
zkServer = new EmbeddedZookeeper(zkConnect);
zkClient =
new ZkClient(zkServer.connectString(), 30000, 30000, ZKStringSerializer$.MODULE$);
brokerPort = TestUtils.choosePort();
logger.info("{}", brokerPort);;
Properties props = TestUtils.createBrokerConfig(BROKER_ID, brokerPort, true);
// props.setProperty("zookeeper.connect", String.valueOf(zkPort));
props.setProperty("zookeeper.session.timeout.ms", "30000");
props.setProperty("zookeeper.connection.timeout.ms", "30000");
logger.info("{}", props);
KafkaConfig config = new KafkaConfig(props);
Time mock = new MockTime();
kafkaServer = TestUtils.createServer(config, mock);
}
开发者ID:trustedanalytics，项目名称:data-acquisition，代码行数:18，代码来源:KafkaEmbeded.java
示例3: startServer
import kafka.utils.TestZKUtils; //导入依赖的package包/类
public static void startServer() throws RuntimeException {
if ( serverStarted && serverClosed ) {
throw new RuntimeException("Kafka test server has already been closed. Cannot generate Kafka server twice.");
}
if (!serverStarted) {
serverStarted = true;
zkConnect = TestZKUtils.zookeeperConnect();
zkServer = new EmbeddedZookeeper(zkConnect);
zkClient = new ZkClient(zkServer.connectString(), 30000, 30000, ZKStringSerializer$.MODULE$);
kafkaPort = TestUtils.choosePort();
Properties props = TestUtils.createBrokerConfig(brokerId, kafkaPort, true);
KafkaConfig config = new KafkaConfig(props);
Time mock = new MockTime();
kafkaServer = TestUtils.createServer(config, mock);
}
}
开发者ID:Hanmourang，项目名称:Gobblin，代码行数:19，代码来源:KafkaTestBase.java
示例4: startServer
import kafka.utils.TestZKUtils; //导入依赖的package包/类
public static void startServer() throws RuntimeException {
if (serverStarted && serverClosed) {
throw new RuntimeException("Kafka test server has already been closed. Cannot generate Kafka server twice.");
}
if (!serverStarted) {
serverStarted = true;
zkConnect = TestZKUtils.zookeeperConnect();
zkServer = new EmbeddedZookeeper(zkConnect);
zkClient = new ZkClient(zkServer.connectString(), 30000, 30000, ZKStringSerializer$.MODULE$);
kafkaPort = TestUtils.choosePort();
Properties props = TestUtils.createBrokerConfig(brokerId, kafkaPort, true);
KafkaConfig config = new KafkaConfig(props);
Time mock = new MockTime();
kafkaServer = TestUtils.createServer(config, mock);
}
}
开发者ID:apache，项目名称:incubator-gobblin，代码行数:19，代码来源:KafkaTestBase.java
示例5: setup
import kafka.utils.TestZKUtils; //导入依赖的package包/类
@BeforeClass
public static void setup() {
zkServer = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect());
zkClient = new ZkClient(zkServer.connectString(), 30000, 30000, ZKStringSerializer$.MODULE$);
kafkaServer = TestUtils.createServer(new KafkaConfig(TestUtils.createBrokerConfig(0, brokerPort, true)),
new MockTime());
kafkaServers = new ArrayList<>();
kafkaServers.add(kafkaServer);
}
开发者ID:FreshetDMS，项目名称:Freshet-Deprecated，代码行数:10，代码来源:AbstractSamzaTest.java
示例6: startZookeeper
import kafka.utils.TestZKUtils; //导入依赖的package包/类
public void startZookeeper() {
zkConnect = TestZKUtils.zookeeperConnect();
try {
zkServer = new EmbeddedZookeeper(zkConnect);
} catch (Exception ex) {
String msg = Utils.format("Error starting zookeeper {}: {}", zkConnect, ex);
throw new RuntimeException(msg, ex);
}
zkClient = new ZkClient(zkServer.connectString(), 30000, 30000, ZKStringSerializer$.MODULE$);
}
开发者ID:streamsets，项目名称:datacollector，代码行数:11，代码来源:KafkaTestUtil08.java
示例7: createBrokerConfig
import kafka.utils.TestZKUtils; //导入依赖的package包/类
private static Properties createBrokerConfig(int i, int port) {
Properties props = new Properties();
props.put("broker.id", String.valueOf(i));
props.put("host.name", "localhost");
props.put("port", String.valueOf(port));
props.put("log.dir", TestUtils.tempDir().getAbsolutePath());
props.put("zookeeper.connect", TestZKUtils.zookeeperConnect());
props.put("replica.socket.timeout.ms", "1500");
props.put("controlled.shutdown.enable", "true");
return props;
}
开发者ID:streamsets，项目名称:datacollector，代码行数:12，代码来源:KafkaTestUtil08.java
示例8: startZookeeper
import kafka.utils.TestZKUtils; //导入依赖的package包/类
public static void startZookeeper() {
zkConnect = TestZKUtils.zookeeperConnect();
try {
zkServer = new EmbeddedZookeeper(zkConnect);
} catch (Exception ex) {
String msg = Utils.format("Error starting zookeeper {}: {}", zkConnect, ex);
throw new RuntimeException(msg, ex);
}
zkClient = new ZkClient(zkServer.connectString(), 30000, 30000, ZKStringSerializer$.MODULE$);
}
开发者ID:streamsets，项目名称:datacollector，代码行数:11，代码来源:KafkaTestUtil.java
示例9: testKafkaTransport
import kafka.utils.TestZKUtils; //导入依赖的package包/类
@Test
public void testKafkaTransport() throws Exception {
String topic = "zipkin";
// Kafka setup
EmbeddedZookeeper zkServer = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect());
ZkClient zkClient = new ZkClient(zkServer.connectString(), 30000, 30000, ZKStringSerializer$.MODULE$);
Properties props = TestUtils.createBrokerConfig(0, TestUtils.choosePort(), false);
KafkaConfig config = new KafkaConfig(props);
KafkaServer kafkaServer = TestUtils.createServer(config, new MockTime());
Buffer<KafkaServer> servers = JavaConversions.asScalaBuffer(Collections.singletonList(kafkaServer));
TestUtils.createTopic(zkClient, topic, 1, 1, servers, new Properties());
zkClient.close();
TestUtils.waitUntilMetadataIsPropagated(servers, topic, 0, 5000);
// HTrace
HTraceConfiguration hTraceConfiguration = HTraceConfiguration.fromKeyValuePairs(
"sampler.classes", "AlwaysSampler",
"span.receiver.classes", ZipkinSpanReceiver.class.getName(),
"zipkin.kafka.metadata.broker.list", config.advertisedHostName() + ":" + config.advertisedPort(),
"zipkin.kafka.topic", topic,
ZipkinSpanReceiver.TRANSPORT_CLASS_KEY, KafkaTransport.class.getName()
);
final Tracer tracer = new Tracer.Builder("test-tracer")
.tracerPool(new TracerPool("test-tracer-pool"))
.conf(hTraceConfiguration)
.build();
String scopeName = "test-kafka-transport-scope";
TraceScope traceScope = tracer.newScope(scopeName);
traceScope.close();
tracer.close();
// Kafka consumer
Properties consumerProps = new Properties();
consumerProps.put("zookeeper.connect", props.getProperty("zookeeper.connect"));
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "testing.group");
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "smallest");
ConsumerConnector connector =
kafka.consumer.Consumer.createJavaConsumerConnector(new kafka.consumer.ConsumerConfig(consumerProps));
Map<String, Integer> topicCountMap = new HashMap<>();
topicCountMap.put(topic, 1);
Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(topicCountMap);
ConsumerIterator<byte[], byte[]> it = streams.get(topic).get(0).iterator();
// Test
Assert.assertTrue("We should have one message in Kafka", it.hasNext());
Span span = new Span();
new TDeserializer(new TBinaryProtocol.Factory()).deserialize(span, it.next().message());
Assert.assertEquals("The span name should match our scope description", span.getName(), scopeName);
kafkaServer.shutdown();
}
开发者ID:apache，项目名称:incubator-htrace，代码行数:57，代码来源:ITZipkinReceiver.java
注：本文中的kafka.utils.TestZKUtils类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论