You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2019/04/04 04:34:29 UTC
[incubator-pinot] 01/01: Remove stale test
RealtimeTableDataManagerTest
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch remove_old_test
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 18282623c2691148fa4d3c9341b2cb517a07004f
Author: Jackie (Xiaotian) Jiang <xa...@linkedin.com>
AuthorDate: Wed Apr 3 21:30:05 2019 -0700
Remove stale test RealtimeTableDataManagerTest
RealtimeTableDataManagerTest is a stale test class and does not have any test enabled
The test itself does not reflect to the data file, and the functionality is covered by the RealtimeClusterIntegrationTest
---
.../realtime/RealtimeTableDataManagerTest.java | 320 ---------------------
1 file changed, 320 deletions(-)
diff --git a/pinot-server/src/test/java/org/apache/pinot/server/integration/realtime/RealtimeTableDataManagerTest.java b/pinot-server/src/test/java/org/apache/pinot/server/integration/realtime/RealtimeTableDataManagerTest.java
deleted file mode 100644
index 76501ce..0000000
--- a/pinot-server/src/test/java/org/apache/pinot/server/integration/realtime/RealtimeTableDataManagerTest.java
+++ /dev/null
@@ -1,320 +0,0 @@
-/**
- * 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.pinot.server.integration.realtime;
-
-import com.yammer.metrics.core.MetricsRegistry;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.TimerTask;
-import java.util.concurrent.TimeUnit;
-import org.apache.commons.configuration.Configuration;
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.PropertiesConfiguration;
-import org.apache.helix.ZNRecord;
-import org.apache.pinot.common.config.TableConfig;
-import org.apache.pinot.common.data.FieldSpec;
-import org.apache.pinot.common.data.FieldSpec.FieldType;
-import org.apache.pinot.common.data.Schema;
-import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
-import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
-import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.common.segment.ReadMode;
-import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.CommonConstants.Segment.Realtime.Status;
-import org.apache.pinot.common.utils.CommonConstants.Segment.SegmentType;
-import org.apache.pinot.core.common.Block;
-import org.apache.pinot.core.common.BlockMetadata;
-import org.apache.pinot.core.common.BlockMultiValIterator;
-import org.apache.pinot.core.common.BlockSingleValIterator;
-import org.apache.pinot.core.common.BlockValSet;
-import org.apache.pinot.core.common.Constants;
-import org.apache.pinot.core.common.DataSource;
-import org.apache.pinot.core.data.manager.config.InstanceDataManagerConfig;
-import org.apache.pinot.core.data.manager.config.TableDataManagerConfig;
-import org.apache.pinot.core.data.manager.realtime.HLRealtimeSegmentDataManager;
-import org.apache.pinot.core.data.manager.realtime.TimerService;
-import org.apache.pinot.core.indexsegment.mutable.MutableSegment;
-import org.apache.pinot.core.indexsegment.mutable.MutableSegmentImplTest;
-import org.apache.pinot.core.segment.index.loader.IndexLoadingConfig;
-import org.apache.pinot.segments.v1.creator.SegmentTestUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testng.annotations.BeforeClass;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-
-public class RealtimeTableDataManagerTest {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeTableDataManagerTest.class);
-
- private static TableConfig tableConfig;
-
- private static InstanceZKMetadata instanceZKMetadata;
- private static RealtimeSegmentZKMetadata realtimeSegmentZKMetadata;
- private static TableDataManagerConfig tableDataManagerConfig;
- private static final String AVRO_DATA = "data/test_data-mv.avro";
- private static String filePath;
- private static Map<String, FieldType> fieldTypeMap;
-
- private static final String TABLE_DATA_MANAGER_NUM_QUERY_EXECUTOR_THREADS = "numQueryExecutorThreads";
- private static final String TABLE_DATA_MANAGER_TYPE = "dataManagerType";
- private static final String READ_MODE = "readMode";
- private static final String TABLE_DATA_MANAGER_DATA_DIRECTORY = "directory";
- private static final String TABLE_DATA_MANAGER_NAME = "name";
-
- private static final long SEGMENT_CONSUMING_TIME = 1000 * 60 * 3;
-
- private static volatile boolean keepOnRunning = true;
-
- @BeforeClass
- public static void setup()
- throws Exception {
- instanceZKMetadata = getInstanceZKMetadata();
- realtimeSegmentZKMetadata = getRealtimeSegmentZKMetadata();
- tableDataManagerConfig = getTableDataManagerConfig();
-
- Map<String, String> streamConfigs = new HashMap<>();
- streamConfigs.put("streamType", "kafka");
- streamConfigs.put("stream.kafka.consumer.type", "highLevel");
- streamConfigs.put("stream.kafka.topic.name", "kafkaTopic");
- streamConfigs
- .put("stream.kafka.decoder.class.name", "org.apache.pinot.core.realtime.impl.kafka.KafkaAvroMessageDecoder");
- streamConfigs.put("stream.kafka.hlc.zk.connect.string", "localhost:1111/zkConnect");
- streamConfigs.put("stream.kafka.decoder.prop.schema.registry.rest.url", "http://localhost:2222/schemaRegistry");
- tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.REALTIME).setTableName("mirror")
- .setStreamConfigs(streamConfigs).build();
- }
-
- private static TableDataManagerConfig getTableDataManagerConfig()
- throws ConfigurationException {
- String tableName = "testTable_R";
- Configuration defaultConfig = new PropertiesConfiguration();
- defaultConfig.addProperty(TABLE_DATA_MANAGER_NAME, tableName);
- String dataDir = "/tmp/" + tableName;
- defaultConfig.addProperty(TABLE_DATA_MANAGER_DATA_DIRECTORY, dataDir);
- defaultConfig.addProperty(READ_MODE, ReadMode.heap.toString());
- defaultConfig.addProperty(TABLE_DATA_MANAGER_NUM_QUERY_EXECUTOR_THREADS, 20);
- TableDataManagerConfig tableDataManagerConfig = new TableDataManagerConfig(defaultConfig);
-
- defaultConfig.addProperty(TABLE_DATA_MANAGER_TYPE, "realtime");
-
- return tableDataManagerConfig;
- }
-
- private InstanceDataManagerConfig makeInstanceDataManagerConfig() {
- InstanceDataManagerConfig dataManagerConfig = mock(InstanceDataManagerConfig.class);
- when(dataManagerConfig.getReadMode()).thenReturn(null);
- when(dataManagerConfig.getAvgMultiValueCount()).thenReturn(null);
- when(dataManagerConfig.getSegmentFormatVersion()).thenReturn(null);
- when(dataManagerConfig.isEnableDefaultColumns()).thenReturn(false);
- when(dataManagerConfig.isEnableSplitCommit()).thenReturn(false);
- when(dataManagerConfig.isRealtimeOffHeapAllocation()).thenReturn(false);
- return dataManagerConfig;
- }
-
- public void testSetup()
- throws Exception {
- InstanceDataManagerConfig dataManagerConfig = makeInstanceDataManagerConfig();
- final HLRealtimeSegmentDataManager manager =
- new HLRealtimeSegmentDataManager(realtimeSegmentZKMetadata, tableConfig, instanceZKMetadata, null,
- tableDataManagerConfig.getDataDir(), new IndexLoadingConfig(dataManagerConfig, tableConfig),
- getTestSchema(), new ServerMetrics(new MetricsRegistry()));
-
- final long start = System.currentTimeMillis();
- TimerService.timer.scheduleAtFixedRate(new TimerTask() {
-
- @Override
- public void run() {
- if (System.currentTimeMillis() - start >= (SEGMENT_CONSUMING_TIME)) {
- keepOnRunning = false;
- }
- }
- }, 1000, 1000 * 60 * 1);
-
- TimerService.timer.scheduleAtFixedRate(new TimerTask() {
-
- @Override
- public void run() {
- long start = System.currentTimeMillis();
- long sum = 0;
- try {
- MutableSegment segment = (MutableSegment) manager.getSegment();
- DataSource mDs = segment.getDataSource("count");
- BlockValSet valSet = mDs.nextBlock().getBlockValueSet();
- BlockSingleValIterator valIt = (BlockSingleValIterator) valSet.iterator();
- int val = valIt.nextIntVal();
- while (val != Constants.EOF) {
- val = valIt.nextIntVal();
- sum += val;
- }
- } catch (Exception e) {
- LOGGER.info("count column exception");
- e.printStackTrace();
- }
-
- long stop = System.currentTimeMillis();
- LOGGER.info("time to scan metric col count : " + (stop - start) + " sum : " + sum);
- }
- }, 20000, 1000 * 5);
-
- TimerService.timer.scheduleAtFixedRate(new TimerTask() {
-
- @Override
- public void run() {
- long start = System.currentTimeMillis();
- long sum = 0;
- try {
- MutableSegment segment = (MutableSegment) manager.getSegment();
- DataSource mDs = segment.getDataSource("viewerId");
- BlockValSet valSet = mDs.nextBlock().getBlockValueSet();
- BlockSingleValIterator valIt = (BlockSingleValIterator) valSet.iterator();
- int val = valIt.nextIntVal();
- while (val != Constants.EOF) {
- val = valIt.nextIntVal();
- sum += val;
- }
- } catch (Exception e) {
- LOGGER.info("viewerId column exception");
- e.printStackTrace();
- }
-
- long stop = System.currentTimeMillis();
- LOGGER.info("time to scan SV dimension col viewerId : " + (stop - start) + " sum : " + sum);
- }
- }, 20000, 1000 * 5);
-
- TimerService.timer.scheduleAtFixedRate(new TimerTask() {
-
- @Override
- public void run() {
- long start = System.currentTimeMillis();
- long sum = 0;
- try {
- MutableSegment segment = (MutableSegment) manager.getSegment();
- DataSource mDs = segment.getDataSource("daysSinceEpoch");
- BlockValSet valSet = mDs.nextBlock().getBlockValueSet();
- BlockSingleValIterator valIt = (BlockSingleValIterator) valSet.iterator();
- int val = valIt.nextIntVal();
- while (val != Constants.EOF) {
- val = valIt.nextIntVal();
- sum += val;
- }
- } catch (Exception e) {
- LOGGER.info("daysSinceEpoch column exception");
- e.printStackTrace();
- }
- long stop = System.currentTimeMillis();
- LOGGER.info("time to scan SV time col daysSinceEpoch : " + (stop - start) + " sum : " + sum);
- }
- }, 20000, 1000 * 5);
-
- TimerService.timer.scheduleAtFixedRate(new TimerTask() {
-
- @Override
- public void run() {
- long start = System.currentTimeMillis();
- long sum = 0;
- float sumOfLengths = 0F;
- float counter = 0F;
- try {
- MutableSegment segment = (MutableSegment) manager.getSegment();
- DataSource mDs = segment.getDataSource("viewerCompanies");
- Block b = mDs.nextBlock();
- BlockValSet valSet = b.getBlockValueSet();
- BlockMultiValIterator valIt = (BlockMultiValIterator) valSet.iterator();
- BlockMetadata m = b.getMetadata();
- int maxVams = m.getMaxNumberOfMultiValues();
- while (valIt.hasNext()) {
- int[] vals = new int[maxVams];
- int len = valIt.nextIntVal(vals);
- for (int i = 0; i < len; i++) {
- sum += vals[i];
- }
- sumOfLengths += len;
- counter++;
- }
- } catch (Exception e) {
- LOGGER.info("daysSinceEpoch column exception");
- e.printStackTrace();
- }
- long stop = System.currentTimeMillis();
- LOGGER.info("time to scan MV col viewerCompanies : " + (stop - start) + " sum : " + sum + " average len : " + (
- sumOfLengths / counter));
- }
- }, 20000, 1000 * 5);
-
- while (keepOnRunning) {
- // Wait for keepOnRunning to be set to false
- }
- }
-
- private static InstanceZKMetadata getInstanceZKMetadata() {
- ZNRecord record = new ZNRecord("Server_localhost_1234");
- Map<String, String> groupIdMap = new HashMap<>();
- Map<String, String> partitionMap = new HashMap<>();
-
- groupIdMap.put("mirror", "groupId_testTable_" + String.valueOf(System.currentTimeMillis()));
- partitionMap.put("testTable_R", "0");
- record.setMapField("KAFKA_HLC_GROUP_MAP", groupIdMap);
- record.setMapField("KAFKA_HLC_PARTITION_MAP", partitionMap);
- return new InstanceZKMetadata(record);
- }
-
- private static RealtimeSegmentZKMetadata getRealtimeSegmentZKMetadata() {
- RealtimeSegmentZKMetadata realtimeSegmentMetadata = new RealtimeSegmentZKMetadata();
- realtimeSegmentMetadata.setSegmentName("testTable_R_1000_groupId0_part0");
- realtimeSegmentMetadata.setTableName("testTable");
- realtimeSegmentMetadata.setSegmentType(SegmentType.REALTIME);
- realtimeSegmentMetadata.setIndexVersion("v1");
- realtimeSegmentMetadata.setStartTime(1000);
- realtimeSegmentMetadata.setEndTime(-1);
- realtimeSegmentMetadata.setTimeUnit(TimeUnit.HOURS);
- realtimeSegmentMetadata.setStatus(Status.IN_PROGRESS);
- realtimeSegmentMetadata.setTotalRawDocs(-1);
- realtimeSegmentMetadata.setCrc(-1);
- realtimeSegmentMetadata.setCreationTime(1000);
- return realtimeSegmentMetadata;
- }
-
- private static Schema getTestSchema()
- throws FileNotFoundException, IOException {
- filePath = MutableSegmentImplTest.class.getClassLoader().getResource(AVRO_DATA).getFile();
- fieldTypeMap = new HashMap<String, FieldSpec.FieldType>();
- fieldTypeMap.put("viewerId", FieldType.DIMENSION);
- fieldTypeMap.put("vieweeId", FieldType.DIMENSION);
- fieldTypeMap.put("viewerPrivacySetting", FieldType.DIMENSION);
- fieldTypeMap.put("vieweePrivacySetting", FieldType.DIMENSION);
- fieldTypeMap.put("viewerObfuscationType", FieldType.DIMENSION);
- fieldTypeMap.put("viewerCompanies", FieldType.DIMENSION);
- fieldTypeMap.put("viewerOccupations", FieldType.DIMENSION);
- fieldTypeMap.put("viewerRegionCode", FieldType.DIMENSION);
- fieldTypeMap.put("viewerIndustry", FieldType.DIMENSION);
- fieldTypeMap.put("viewerSchool", FieldType.DIMENSION);
- fieldTypeMap.put("weeksSinceEpochSunday", FieldType.DIMENSION);
- fieldTypeMap.put("daysSinceEpoch", FieldType.DIMENSION);
- fieldTypeMap.put("minutesSinceEpoch", FieldType.TIME);
- fieldTypeMap.put("count", FieldType.METRIC);
- return SegmentTestUtils.extractSchemaFromAvro(new File(filePath), fieldTypeMap, TimeUnit.MINUTES);
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org