apache / carbondata / 681

Build # 681

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/c21286618656d4dfca36ba10ea4412f640903eea">c21286618<a href="https://github.com/apache/carbondata/commit/c21286618656d4dfca36ba10ea4412f640903eea">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/fd9d6e1cc5a9b39581faa314393697ad12dd4ffb">fd9d6e1cc</a>">fd9d6e1cc</a><a href="https://github.com/apache/carbondata/commit/c21286618656d4dfca36ba10ea4412f640903eea"> into f2398948c">f2398948c</a>

Run Details

4 of 4 new or added lines in 2 files covered. (100.0%)

59909 of 74397 relevant lines covered (80.53%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.95

/core/src/main/java/org/apache/carbondata/core/dictionary/server/NonSecureDictionaryServer.java

/*
 * 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.carbondata.core.dictionary.server;

import java.net.InetSocketAddress;

import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessage;
import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessageType;
import org.apache.carbondata.core.dictionary.service.AbstractDictionaryServer;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.util.CarbonProperties;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

/**
 * Dictionary Server to generate dictionary keys.
 */
public class NonSecureDictionaryServer extends AbstractDictionaryServer
    implements DictionaryServer {

  private static final LogService LOGGER =
      LogServiceFactory.getLogService(NonSecureDictionaryServer.class.getName());

  private NonSecureDictionaryServerHandler nonSecureDictionaryServerHandler;

  private EventLoopGroup boss;
  private EventLoopGroup worker;
  private int port;
  private String host;
  private static Object lock = new Object();
  private static NonSecureDictionaryServer INSTANCE = null;

  private NonSecureDictionaryServer(int port) {
    this.port = port;
    startServer();
  }

  public static synchronized DictionaryServer getInstance(int port, CarbonTable carbonTable)
      throws Exception {
    if (INSTANCE == null) {
      INSTANCE = new NonSecureDictionaryServer(port);
    }
    INSTANCE.initializeDictionaryGenerator(carbonTable);
    return INSTANCE;
  }

  /**
   * start dictionary server
   *
   */
  @Override public void startServer() {
    LOGGER.info("Starting Dictionary Server in Non Secure Mode");
    nonSecureDictionaryServerHandler = new NonSecureDictionaryServerHandler();
    String workerThreads = CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.DICTIONARY_WORKER_THREADS,
            CarbonCommonConstants.DICTIONARY_WORKER_THREADS_DEFAULT);
    boss = new NioEventLoopGroup(1);
    worker = new NioEventLoopGroup(Integer.parseInt(workerThreads));
    // Configure the server.
    bindToPort();
  }

  /**
   * Binds dictionary server to an available port.
   *
   */
  @Override public void bindToPort() {
    long start = System.currentTimeMillis();
    // Configure the server.
    int i = 0;
    while (i < 10) {
      int newPort = port + i;
      try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(boss, worker);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
          @Override public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline
                .addLast("LengthDecoder", new LengthFieldBasedFrameDecoder(1048576, 0, 2, 0, 2));
            pipeline.addLast("NonSecureDictionaryServerHandler", nonSecureDictionaryServerHandler);
          }
        });
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        String hostToBind = findLocalIpAddress(LOGGER);
        //iteratively listening to newports
        InetSocketAddress address = hostToBind == null ?
            new InetSocketAddress(newPort) :
            new InetSocketAddress(hostToBind, newPort);
        bootstrap.bind(address).sync();
        LOGGER.info("Dictionary Server started, Time spent " + (System.currentTimeMillis() - start)
            + " Listening on port " + newPort);
        this.port = newPort;
        this.host = hostToBind;
        break;
      } catch (Exception e) {
        LOGGER.error(e, "Dictionary Server Failed to bind to port:" + newPort);
        if (i == 9) {
          throw new RuntimeException("Dictionary Server Could not bind to any port");
        }
      }
      i++;
    }
  }

  /**
   * @return Port on which the NonSecureDictionaryServer has started.
   */
  @Override public int getPort() {
    return port;
  }

  @Override public String getSecretKey() {
    return null;
  }

  @Override public boolean isEncryptSecureServer() {
    return false;
  }

  @Override public String getHost() {
    return host;
  }

  /**
   * shutdown dictionary server
   *
   * @throws Exception
   */
  @Override public void shutdown() throws Exception {
    LOGGER.info("Shutting down dictionary server");
    worker.shutdownGracefully();
    boss.shutdownGracefully();
  }

  /**
   * Write Dictionary for one table.
   *
   * @throws Exception
   */

  @Override
  public void writeTableDictionary(String uniqueTableName) throws Exception {
    DictionaryMessage key = new DictionaryMessage();
    key.setTableUniqueId(uniqueTableName);
    key.setType(DictionaryMessageType.WRITE_TABLE_DICTIONARY);
    nonSecureDictionaryServerHandler.processMessage(key);
  }

  public void initializeDictionaryGenerator(CarbonTable carbonTable) throws Exception {
    nonSecureDictionaryServerHandler.initializeTable(carbonTable);
  }

}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17	package org.apache.carbondata.core.dictionary.server;
18
19	import java.net.InetSocketAddress;
20
21	import org.apache.carbondata.common.logging.LogService;
22	import org.apache.carbondata.common.logging.LogServiceFactory;
23	import org.apache.carbondata.core.constants.CarbonCommonConstants;
24	import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessage;
25	import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessageType;
26	import org.apache.carbondata.core.dictionary.service.AbstractDictionaryServer;
27	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
28	import org.apache.carbondata.core.util.CarbonProperties;
29
30	import io.netty.bootstrap.ServerBootstrap;
31	import io.netty.channel.ChannelInitializer;
32	import io.netty.channel.ChannelOption;
33	import io.netty.channel.ChannelPipeline;
34	import io.netty.channel.EventLoopGroup;
35	import io.netty.channel.nio.NioEventLoopGroup;
36	import io.netty.channel.socket.SocketChannel;
37	import io.netty.channel.socket.nio.NioServerSocketChannel;
38	import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
39
40	/**
41	* Dictionary Server to generate dictionary keys.
42	*/
43	public class NonSecureDictionaryServer extends AbstractDictionaryServer
44	implements DictionaryServer {
45
46	private static final LogService LOGGER =	2✔
47	LogServiceFactory.getLogService(NonSecureDictionaryServer.class.getName());	2✔
48
49	private NonSecureDictionaryServerHandler nonSecureDictionaryServerHandler;
50
51	private EventLoopGroup boss;
52	private EventLoopGroup worker;
53	private int port;
54	private String host;
55	private static Object lock = new Object();	2✔
56	private static NonSecureDictionaryServer INSTANCE = null;	2✔
57
58	private NonSecureDictionaryServer(int port) {	2✔
59	this.port = port;	2✔
60	startServer();	2✔
61	}	2✔
62
63	public static synchronized DictionaryServer getInstance(int port, CarbonTable carbonTable)
64	throws Exception {
65	if (INSTANCE == null) {	2✔
66	INSTANCE = new NonSecureDictionaryServer(port);	2✔
67	}
68	INSTANCE.initializeDictionaryGenerator(carbonTable);	2✔
69	return INSTANCE;	2✔
70	}
71
72	/**
73	* start dictionary server
74	*
75	*/
76	@Override public void startServer() {
77	LOGGER.info("Starting Dictionary Server in Non Secure Mode");	2✔
78	nonSecureDictionaryServerHandler = new NonSecureDictionaryServerHandler();	2✔
79	String workerThreads = CarbonProperties.getInstance()	2✔
80	.getProperty(CarbonCommonConstants.DICTIONARY_WORKER_THREADS,	2✔
81	CarbonCommonConstants.DICTIONARY_WORKER_THREADS_DEFAULT);
82	boss = new NioEventLoopGroup(1);	2✔
83	worker = new NioEventLoopGroup(Integer.parseInt(workerThreads));	2✔
84	// Configure the server.
85	bindToPort();	2✔
86	}	2✔
87
88	/**
89	* Binds dictionary server to an available port.
90	*
91	*/
92	@Override public void bindToPort() {
93	long start = System.currentTimeMillis();	2✔
94	// Configure the server.
95	int i = 0;	2✔
96	while (i < 10) {	2✔
97	int newPort = port + i;	2✔
98	try {
99	ServerBootstrap bootstrap = new ServerBootstrap();	2✔
100	bootstrap.group(boss, worker);	2✔
101	bootstrap.channel(NioServerSocketChannel.class);	2✔
102	bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {	2✔
103	@Override public void initChannel(SocketChannel ch) throws Exception {
104	ChannelPipeline pipeline = ch.pipeline();	2✔
105	pipeline	2✔
106	.addLast("LengthDecoder", new LengthFieldBasedFrameDecoder(1048576, 0, 2, 0, 2));	2✔
107	pipeline.addLast("NonSecureDictionaryServerHandler", nonSecureDictionaryServerHandler);	2✔
108	}	2✔
109	});
110	bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);	2✔
111	String hostToBind = findLocalIpAddress(LOGGER);	2✔
112	//iteratively listening to newports
113	InetSocketAddress address = hostToBind == null ?	2✔
114	new InetSocketAddress(newPort) :
115	new InetSocketAddress(hostToBind, newPort);
116	bootstrap.bind(address).sync();	2✔
117	LOGGER.info("Dictionary Server started, Time spent " + (System.currentTimeMillis() - start)	2✔
118	+ " Listening on port " + newPort);
119	this.port = newPort;	2✔
120	this.host = hostToBind;	2✔
121	break;	2✔
122	} catch (Exception e) {	×
123	LOGGER.error(e, "Dictionary Server Failed to bind to port:" + newPort);	×
124	if (i == 9) {	×
125	throw new RuntimeException("Dictionary Server Could not bind to any port");	×
126	}
127	}
128	i++;	×
129	}	×
130	}	2✔
131
132	/**
133	* @return Port on which the NonSecureDictionaryServer has started.
134	*/
135	@Override public int getPort() {
136	return port;	1✔
137	}
138
139	@Override public String getSecretKey() {
140	return null;	×
141	}
142
143	@Override public boolean isEncryptSecureServer() {
144	return false;	×
145	}
146
147	@Override public String getHost() {
148	return host;	2✔
149	}
150
151	/**
152	* shutdown dictionary server
153	*
154	* @throws Exception
155	*/
156	@Override public void shutdown() throws Exception {
157	LOGGER.info("Shutting down dictionary server");	×
158	worker.shutdownGracefully();	×
159	boss.shutdownGracefully();	×
160	}	×
161
162	/**
163	* Write Dictionary for one table.
164	*
165	* @throws Exception
166	*/
167
168	@Override
169	public void writeTableDictionary(String uniqueTableName) throws Exception {
170	DictionaryMessage key = new DictionaryMessage();	1✔
171	key.setTableUniqueId(uniqueTableName);	1✔
172	key.setType(DictionaryMessageType.WRITE_TABLE_DICTIONARY);	1✔
173	nonSecureDictionaryServerHandler.processMessage(key);	1✔
174	}	1✔
175
176	public void initializeDictionaryGenerator(CarbonTable carbonTable) throws Exception {
177	nonSecureDictionaryServerHandler.initializeTable(carbonTable);	2✔
178	}	2✔
179
180	}

apache / carbondata / 681

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous