[GitHub] [incubator-pinot] codecov-io edited a comment on issue #4218: Add RealtimeConsumptionCatchupServiceCallback
codecov-io edited a comment on issue #4218: Add RealtimeConsumptionCatchupServiceCallback URL: https://github.com/apache/incubator-pinot/pull/4218#issuecomment-493596443 # [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/4218?src=pr=h1) Report > Merging [#4218](https://codecov.io/gh/apache/incubator-pinot/pull/4218?src=pr=desc) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/45a8430bc3d981b44e2d8e242c75480d7dc593e6?src=pr=desc) will **decrease** coverage by `0.02%`. > The diff coverage is `35.71%`. [](https://codecov.io/gh/apache/incubator-pinot/pull/4218?src=pr=tree) ```diff @@ Coverage Diff @@ ## master#4218 +/- ## - Coverage 67.24% 67.21% -0.03% Complexity 20 20 Files 1041 1041 Lines 5151451546 +32 Branches 7216 7226 +10 + Hits 3464034648 +8 - Misses1450614533 +27 + Partials 2368 2365 -3 ``` | [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/4218?src=pr=tree) | Coverage Δ | Complexity Δ | | |---|---|---|---| | [...org/apache/pinot/common/utils/CommonConstants.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdXRpbHMvQ29tbW9uQ29uc3RhbnRzLmphdmE=) | `45.65% <ø> (ø)` | `0 <0> (ø)` | :arrow_down: | | [...g/apache/pinot/common/config/TableNameBuilder.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vY29uZmlnL1RhYmxlTmFtZUJ1aWxkZXIuamF2YQ==) | `92.3% <0%> (-3.7%)` | `0 <0> (ø)` | | | [...pinot/server/starter/helix/HelixServerStarter.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvc3RhcnRlci9oZWxpeC9IZWxpeFNlcnZlclN0YXJ0ZXIuamF2YQ==) | `45.88% <100%> (ø)` | `0 <0> (ø)` | :arrow_down: | | [.../pinot/broker/broker/helix/HelixBrokerStarter.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0hlbGl4QnJva2VyU3RhcnRlci5qYXZh) | `75.53% <100%> (-0.18%)` | `0 <0> (ø)` | | | [...a/org/apache/pinot/common/utils/ServiceStatus.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdXRpbHMvU2VydmljZVN0YXR1cy5qYXZh) | `69.58% <32.69%> (-14.37%)` | `0 <0> (ø)` | | | [...a/manager/realtime/RealtimeSegmentDataManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvUmVhbHRpbWVTZWdtZW50RGF0YU1hbmFnZXIuamF2YQ==) | `50% <0%> (-50%)` | `0% <0%> (ø)` | | | [...r/validation/RealtimeSegmentValidationManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci92YWxpZGF0aW9uL1JlYWx0aW1lU2VnbWVudFZhbGlkYXRpb25NYW5hZ2VyLmphdmE=) | `43.33% <0%> (-36.67%)` | `0% <0%> (ø)` | | | [...apache/pinot/common/metrics/ValidationMetrics.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9WYWxpZGF0aW9uTWV0cmljcy5qYXZh) | `20.28% <0%> (-23.19%)` | `0% <0%> (ø)` | | | [.../impl/dictionary/LongOffHeapMutableDictionary.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9yZWFsdGltZS9pbXBsL2RpY3Rpb25hcnkvTG9uZ09mZkhlYXBNdXRhYmxlRGljdGlvbmFyeS5qYXZh) | `83.92% <0%> (-10.72%)` | `0% <0%> (ø)` | | | [...impl/dictionary/FloatOffHeapMutableDictionary.java](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9yZWFsdGltZS9pbXBsL2RpY3Rpb25hcnkvRmxvYXRPZmZIZWFwTXV0YWJsZURpY3Rpb25hcnkuamF2YQ==) | `82.14% <0%> (-7.15%)` | `0% <0%> (ø)` | | | ... and [25 more](https://codecov.io/gh/apache/incubator-pinot/pull/4218/diff?src=pr=tree-more) | | -- [Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/4218?src=pr=continue). > **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta) > `Δ =
[GitHub] [incubator-pinot] npawar commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
npawar commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287546381
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionLevelConsumer.java
##
@@ -0,0 +1,121 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.realtime.stream.MessageBatch;
+import org.apache.pinot.core.realtime.stream.PartitionLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamConfig;
+import org.apache.pinot.core.util.AvroUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils.*;
+
+
+/**
+ * Implementation of {@link PartitionLevelConsumer} for fake stream
+ * Unpacks tar files in
/resources/data/On_Time_Performance_2014_partition_.tar.gz as source
of messages
Review comment:
I think that's a little out of scope.. the kafka stream that I'm trying to
replace also read only from files, so we have no tests that need this. We are
not even using the partition level consumer of the fake stream anywhere in our
code yet, so maybe for now we can leave it to hard coded
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] npawar commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
npawar commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287546327
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
##
@@ -0,0 +1,101 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import org.apache.pinot.common.data.Schema;
+import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.core.data.GenericRow;
+import org.apache.pinot.core.realtime.stream.MessageBatch;
+import org.apache.pinot.core.realtime.stream.OffsetCriteria;
+import org.apache.pinot.core.realtime.stream.PartitionLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamConfig;
+import org.apache.pinot.core.realtime.stream.StreamConsumerFactory;
+import org.apache.pinot.core.realtime.stream.StreamConsumerFactoryProvider;
+import org.apache.pinot.core.realtime.stream.StreamDecoderProvider;
+import org.apache.pinot.core.realtime.stream.StreamLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamMessageDecoder;
+import org.apache.pinot.core.realtime.stream.StreamMetadataProvider;
+
+
+/**
+ * Implementation of {@link StreamConsumerFactory} for a fake stream
+ * Data source is /resources/data/fakestream_avro_data.tar.gz
+ * Avro schema is /resources/data/fakestream/fake_stream_avro_schema.avsc
+ * Pinot schema is /resources/data/fakestream/fake_stream_pinot_schema.avsc
+ */
+public class FakeStreamConsumerFactory extends StreamConsumerFactory {
+
+ @Override
+ public PartitionLevelConsumer createPartitionLevelConsumer(String clientId,
int partition) {
+return new FakePartitionLevelConsumer(partition, _streamConfig);
+ }
+
+ @Override
+ public StreamLevelConsumer createStreamLevelConsumer(String clientId, String
tableName, Schema schema,
+ InstanceZKMetadata instanceZKMetadata, ServerMetrics serverMetrics) {
+return new FakeStreamLevelConsumer();
+ }
+
+ @Override
+ public StreamMetadataProvider createPartitionMetadataProvider(String
clientId, int partition) {
+return new FakeStreamMetadataProvider(_streamConfig);
+ }
+
+ @Override
+ public StreamMetadataProvider createStreamMetadataProvider(String clientId) {
+return new FakeStreamMetadataProvider(_streamConfig);
+ }
+
+ public static void main(String[] args) throws Exception {
Review comment:
I've kept it as a demonstration of how to use this fake stream
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] npawar commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
npawar commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287546321
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamLevelConsumer.java
##
@@ -0,0 +1,49 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import org.apache.pinot.core.data.GenericRow;
+import org.apache.pinot.core.realtime.stream.StreamLevelConsumer;
+
+
+/**
+ * Test implementation of {@link StreamLevelConsumer}
+ * This is currently a no-op
+ */
+public class FakeStreamLevelConsumer implements StreamLevelConsumer {
Review comment:
I've mocked that in the partition level consumer. We haven't been using a
stream level consumer with mock data anywhere. Will implement it if we need to
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] npawar commented on issue #4218: Add RealtimeConsumptionCatchupServiceCallback
npawar commented on issue #4218: Add RealtimeConsumptionCatchupServiceCallback URL: https://github.com/apache/incubator-pinot/pull/4218#issuecomment-495823638 Fixed suggestions @sunithabeeram @mcvsubbu I have refactored it such that we are not making the call to getResourcesInCluster and getResourceIdealState in every constructor. It is now happening only in 1 constructor. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request #4240: Skip pipeline validation when disabling it
xiaohui-sun commented on a change in pull request #4240: Skip pipeline
validation when disabling it
URL: https://github.com/apache/incubator-pinot/pull/4240#discussion_r287533490
##
File path:
thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/detection/yaml/YamlResource.java
##
@@ -343,14 +343,21 @@ void updateDetectionPipeline(long detectionID, String
yamlDetectionConfig, long
// Translate config from YAML to detection config (JSON)
TreeMap newDetectionConfigMap = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
newDetectionConfigMap.putAll(ConfigUtils.getMap(this.yaml.load(yamlDetectionConfig)));
-detectionConfig = buildDetectionConfigFromYaml(startTime, endTime,
newDetectionConfigMap, existingDetectionConfig);
-detectionConfig.setYaml(yamlDetectionConfig);
-
-// Validate updated config before saving it
-detectionValidator.validateUpdatedConfig(detectionConfig,
existingDetectionConfig);
-// Save the detection config
-Long id = this.detectionConfigDAO.save(detectionConfig);
-Preconditions.checkNotNull(id, "Error while saving the detection
pipeline");
+// If it is to disable the pipeline then no need to do validation and
parsing.
+// It is possible that the metric or dataset was deleted so the validation
will fail.
+if (!MapUtils.getBooleanValue(newDetectionConfigMap, PROP_ACTIVE, true)) {
+ existingDetectionConfig.setActive(false);
Review comment:
Good point. I put it into "finally" now. Do you have other ideas?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
mcvsubbu commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287519199
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionLevelConsumer.java
##
@@ -0,0 +1,121 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.realtime.stream.MessageBatch;
+import org.apache.pinot.core.realtime.stream.PartitionLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamConfig;
+import org.apache.pinot.core.util.AvroUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils.*;
+
+
+/**
+ * Implementation of {@link PartitionLevelConsumer} for fake stream
+ * Unpacks tar files in
/resources/data/On_Time_Performance_2014_partition_.tar.gz as source
of messages
+ */
+public class FakePartitionLevelConsumer implements PartitionLevelConsumer {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(FakePartitionLevelConsumer.class.getName());
+
+ private List messageOffsets = new ArrayList<>();
+ private List messageBytes = new ArrayList<>();
+
+ FakePartitionLevelConsumer(int partition, StreamConfig streamConfig) {
+
+// TODO: this logic can move to a FakeStreamProducer instead of being
inside the Consumer
+File tempDir = new File(FileUtils.getTempDirectory(),
getClass().getSimpleName());
+File outputDir = new File(tempDir, String.valueOf(partition));
+
+int offset = 0;
+
+try (ByteArrayOutputStream outputStream = new
ByteArrayOutputStream(65536)) {
+ File avroFile = unpackAvroTarFile(outputDir).get(0);
+
+ int numPartitions = FakeStreamConfigUtils.getNumPartitions(streamConfig);
+
+ try (DataFileStream reader =
AvroUtils.getAvroReader(avroFile)) {
+BinaryEncoder binaryEncoder = new
EncoderFactory().directBinaryEncoder(outputStream, null);
+GenericDatumWriter datumWriter = new
GenericDatumWriter<>(reader.getSchema());
+
+int recordNumber = 0;
+for (GenericRecord genericRecord : reader) {
+ if (getPartitionNumber(recordNumber++, numPartitions) != partition) {
+continue;
+ }
+ outputStream.reset();
+
+ datumWriter.write(genericRecord, binaryEncoder);
+ binaryEncoder.flush();
+
+ byte[] bytes = outputStream.toByteArray();
+ // contiguous offsets
+ messageOffsets.add(offset++);
+ messageBytes.add(bytes);
+}
+ }
+} catch (Exception e) {
+ LOGGER.error("Could not create {}",
FakePartitionLevelConsumer.class.getName(), e);
+} finally {
+ FileUtils.deleteQuietly(outputDir);
+}
+ }
+
+ @Override
+ public MessageBatch fetchMessages(long startOffset, long endOffset, int
timeoutMillis) throws TimeoutException {
+if (startOffset >= FakeStreamConfigUtils.getLargestOffset()) {
Review comment:
Thanks for clarifying, @npawar . We cannot do the proposed method since we
want to return records based on an offset.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe,
[GitHub] [incubator-pinot] snleee commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
snleee commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287513134 ## File path: pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java ## @@ -88,7 +85,7 @@ + "\": \"" + maxTimeForSegmentCloseMs + "\", \n" + " \"stream.kafka.broker.list\": \"broker:\", \n" + " \"stream.kafka.consumer.prop.auto.offset.reset\": \"smallest\", \n" Review comment: Do we need `stream.kafka.*` configs for `FakeStreamConsumer`? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
snleee commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287515166
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionLevelConsumer.java
##
@@ -0,0 +1,121 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.realtime.stream.MessageBatch;
+import org.apache.pinot.core.realtime.stream.PartitionLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamConfig;
+import org.apache.pinot.core.util.AvroUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils.*;
+
+
+/**
+ * Implementation of {@link PartitionLevelConsumer} for fake stream
+ * Unpacks tar files in
/resources/data/On_Time_Performance_2014_partition_.tar.gz as source
of messages
Review comment:
It would be great if we can configure the input avro files instead of hard
coding it to a specific file in case we want to reuse this fake consumer but
use different data.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
snleee commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287514421
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamLevelConsumer.java
##
@@ -0,0 +1,49 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import org.apache.pinot.core.data.GenericRow;
+import org.apache.pinot.core.realtime.stream.StreamLevelConsumer;
+
+
+/**
+ * Test implementation of {@link StreamLevelConsumer}
+ * This is currently a no-op
+ */
+public class FakeStreamLevelConsumer implements StreamLevelConsumer {
Review comment:
Why `FakeStreamLevelConsumer` is no-op? I thought that this would provide
some events that you feed as a list of `GenericRow` or avro files?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
snleee commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287513489
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
##
@@ -0,0 +1,101 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import org.apache.pinot.common.data.Schema;
+import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.core.data.GenericRow;
+import org.apache.pinot.core.realtime.stream.MessageBatch;
+import org.apache.pinot.core.realtime.stream.OffsetCriteria;
+import org.apache.pinot.core.realtime.stream.PartitionLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamConfig;
+import org.apache.pinot.core.realtime.stream.StreamConsumerFactory;
+import org.apache.pinot.core.realtime.stream.StreamConsumerFactoryProvider;
+import org.apache.pinot.core.realtime.stream.StreamDecoderProvider;
+import org.apache.pinot.core.realtime.stream.StreamLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamMessageDecoder;
+import org.apache.pinot.core.realtime.stream.StreamMetadataProvider;
+
+
+/**
+ * Implementation of {@link StreamConsumerFactory} for a fake stream
+ * Data source is /resources/data/fakestream_avro_data.tar.gz
+ * Avro schema is /resources/data/fakestream/fake_stream_avro_schema.avsc
+ * Pinot schema is /resources/data/fakestream/fake_stream_pinot_schema.avsc
+ */
+public class FakeStreamConsumerFactory extends StreamConsumerFactory {
+
+ @Override
+ public PartitionLevelConsumer createPartitionLevelConsumer(String clientId,
int partition) {
+return new FakePartitionLevelConsumer(partition, _streamConfig);
+ }
+
+ @Override
+ public StreamLevelConsumer createStreamLevelConsumer(String clientId, String
tableName, Schema schema,
+ InstanceZKMetadata instanceZKMetadata, ServerMetrics serverMetrics) {
+return new FakeStreamLevelConsumer();
+ }
+
+ @Override
+ public StreamMetadataProvider createPartitionMetadataProvider(String
clientId, int partition) {
+return new FakeStreamMetadataProvider(_streamConfig);
+ }
+
+ @Override
+ public StreamMetadataProvider createStreamMetadataProvider(String clientId) {
+return new FakeStreamMetadataProvider(_streamConfig);
+ }
+
+ public static void main(String[] args) throws Exception {
Review comment:
`main` function is required? Can we remove it if it's for testing purpose?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
snleee commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287512796
##
File path:
pinot-core/src/test/resources/data/fakestream/fake_stream_pinot_schema.json
##
@@ -0,0 +1,335 @@
+{
+ "schemaName": "mytable",
+ "dimensionFieldSpecs": [
+{
+ "name": "AirlineID",
+ "dataType": "LONG"
+},
+{
+ "name": "ArrTime",
+ "dataType": "INT"
+},
+{
+ "name": "ArrTimeBlk",
+ "dataType": "STRING"
+},
+{
+ "name": "CRSArrTime",
+ "dataType": "INT"
+},
+{
+ "name": "CRSDepTime",
+ "dataType": "INT"
+},
+{
+ "name": "CRSElapsedTime",
+ "dataType": "INT"
+},
+{
+ "name": "CancellationCode",
+ "dataType": "STRING"
+},
+{
+ "name": "Carrier",
+ "dataType": "STRING"
+},
+{
+ "name": "DayOfWeek",
+ "dataType": "INT"
+},
+{
+ "name": "DayofMonth",
+ "dataType": "INT"
+},
+{
+ "name": "DepTime",
+ "dataType": "INT"
+},
+{
+ "name": "DepTimeBlk",
+ "dataType": "STRING"
+},
+{
+ "name": "Dest",
+ "dataType": "STRING"
+},
+{
+ "name": "DestAirportID",
+ "dataType": "INT"
+},
+{
+ "name": "DestAirportSeqID",
+ "dataType": "INT"
+},
+{
+ "name": "DestCityMarketID",
+ "dataType": "INT"
+},
+{
+ "name": "DestCityName",
+ "dataType": "STRING"
+},
+{
+ "name": "DestState",
+ "dataType": "STRING"
+},
+{
+ "name": "DestStateFips",
+ "dataType": "INT"
+},
+{
+ "name": "DestStateName",
+ "dataType": "STRING"
+},
+{
+ "name": "DestWac",
+ "dataType": "INT"
+},
+{
+ "name": "Distance",
+ "dataType": "INT"
+},
+{
+ "name": "DistanceGroup",
+ "dataType": "INT"
+},
+{
+ "name": "DivActualElapsedTime",
+ "dataType": "INT"
+},
+{
+ "name": "DivAirportIDs",
+ "dataType": "INT",
+ "singleValueField": false
+},
+{
+ "name": "DivAirportLandings",
+ "dataType": "INT"
+},
+{
+ "name": "DivAirportSeqIDs",
+ "dataType": "INT",
+ "singleValueField": false
+},
+{
+ "name": "DivAirports",
+ "dataType": "STRING",
+ "singleValueField": false
+},
+{
+ "name": "DivArrDelay",
+ "dataType": "INT"
+},
+{
+ "name": "DivDistance",
+ "dataType": "INT"
+},
+{
+ "name": "DivLongestGTimes",
+ "dataType": "FLOAT",
+ "singleValueField": false
+},
+{
+ "name": "DivReachedDest",
+ "dataType": "INT"
+},
+{
+ "name": "DivTailNums",
+ "dataType": "STRING",
+ "singleValueField": false
+},
+{
+ "name": "DivTotalGTimes",
+ "dataType": "LONG",
+ "singleValueField": false
+},
+{
+ "name": "DivWheelsOffs",
+ "dataType": "INT",
+ "singleValueField": false
+},
+{
+ "name": "DivWheelsOns",
+ "dataType": "INT",
+ "singleValueField": false
+},
+{
+ "name": "Diverted",
+ "dataType": "INT"
+},
+{
+ "name": "FirstDepTime",
+ "dataType": "INT"
+},
+{
+ "name": "FlightDate",
+ "dataType": "STRING"
+},
+{
+ "name": "FlightNum",
+ "dataType": "INT"
+},
+{
+ "name": "Flights",
+ "dataType": "INT"
+},
+{
+ "name": "LongestAddGTime",
+ "dataType": "INT"
+},
+{
+ "name": "Month",
+ "dataType": "INT"
+},
+{
+ "name": "Origin",
+ "dataType": "STRING"
+},
+{
+ "name": "OriginAirportID",
+ "dataType": "INT"
+},
+{
+ "name": "OriginAirportSeqID",
+ "dataType": "INT"
+},
+{
+ "name": "OriginCityMarketID",
+ "dataType": "INT"
+},
+{
+ "name": "OriginCityName",
+ "dataType": "STRING"
+},
+{
+ "name": "OriginState",
+ "dataType": "STRING"
+},
+{
+ "name": "OriginStateFips",
+ "dataType": "INT"
+},
+{
+ "name": "OriginStateName",
+ "dataType": "STRING"
+},
+{
+ "name": "OriginWac",
+ "dataType": "INT"
+},
+{
+ "name": "Quarter",
+ "dataType": "INT"
+},
+{
+ "name": "RandomAirports",
+ "dataType": "STRING",
+ "singleValueField": false
+},
+{
+ "name": "TailNum",
+ "dataType": "STRING"
+},
+{
+ "name": "TaxiIn",
+ "dataType": "INT"
+},
+{
+ "name": "TaxiOut",
+ "dataType": "INT"
+},
+{
+ "name": "Year",
+ "dataType": "INT"
+},
+{
+ "name": "WheelsOn",
+ "dataType": "INT"
+},
+{
+ "name": "WheelsOff",
+ "dataType": "INT"
+
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request #4240: Skip pipeline validation when disabling it
akshayrai commented on a change in pull request #4240: Skip pipeline validation
when disabling it
URL: https://github.com/apache/incubator-pinot/pull/4240#discussion_r287510920
##
File path:
thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/detection/yaml/YamlResource.java
##
@@ -343,14 +343,21 @@ void updateDetectionPipeline(long detectionID, String
yamlDetectionConfig, long
// Translate config from YAML to detection config (JSON)
TreeMap newDetectionConfigMap = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
newDetectionConfigMap.putAll(ConfigUtils.getMap(this.yaml.load(yamlDetectionConfig)));
-detectionConfig = buildDetectionConfigFromYaml(startTime, endTime,
newDetectionConfigMap, existingDetectionConfig);
-detectionConfig.setYaml(yamlDetectionConfig);
-
-// Validate updated config before saving it
-detectionValidator.validateUpdatedConfig(detectionConfig,
existingDetectionConfig);
-// Save the detection config
-Long id = this.detectionConfigDAO.save(detectionConfig);
-Preconditions.checkNotNull(id, "Error while saving the detection
pipeline");
+// If it is to disable the pipeline then no need to do validation and
parsing.
+// It is possible that the metric or dataset was deleted so the validation
will fail.
+if (!MapUtils.getBooleanValue(newDetectionConfigMap, PROP_ACTIVE, true)) {
+ existingDetectionConfig.setActive(false);
Review comment:
The raw yaml config needs to be injected into the detection object before
saving. Otherwise, the config will still show `active: true`.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun opened a new pull request #4240: Skip pipeline validation when disabling it
xiaohui-sun opened a new pull request #4240: Skip pipeline validation when disabling it URL: https://github.com/apache/incubator-pinot/pull/4240 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
mcvsubbu commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287475250
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionLevelConsumer.java
##
@@ -0,0 +1,121 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.realtime.stream.MessageBatch;
+import org.apache.pinot.core.realtime.stream.PartitionLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamConfig;
+import org.apache.pinot.core.util.AvroUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils.*;
+
+
+/**
+ * Implementation of {@link PartitionLevelConsumer} for fake stream
+ * Unpacks tar files in
/resources/data/On_Time_Performance_2014_partition_.tar.gz as source
of messages
+ */
+public class FakePartitionLevelConsumer implements PartitionLevelConsumer {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(FakePartitionLevelConsumer.class.getName());
+
+ private List messageOffsets = new ArrayList<>();
+ private List messageBytes = new ArrayList<>();
+
+ FakePartitionLevelConsumer(int partition, StreamConfig streamConfig) {
+
+// TODO: this logic can move to a FakeStreamProducer instead of being
inside the Consumer
+File tempDir = new File(FileUtils.getTempDirectory(),
getClass().getSimpleName());
+File outputDir = new File(tempDir, String.valueOf(partition));
+
+int offset = 0;
+
+try (ByteArrayOutputStream outputStream = new
ByteArrayOutputStream(65536)) {
+ File avroFile = unpackAvroTarFile(outputDir).get(0);
+
+ int numPartitions = FakeStreamConfigUtils.getNumPartitions(streamConfig);
+
+ try (DataFileStream reader =
AvroUtils.getAvroReader(avroFile)) {
+BinaryEncoder binaryEncoder = new
EncoderFactory().directBinaryEncoder(outputStream, null);
+GenericDatumWriter datumWriter = new
GenericDatumWriter<>(reader.getSchema());
+
+int recordNumber = 0;
+for (GenericRecord genericRecord : reader) {
+ if (getPartitionNumber(recordNumber++, numPartitions) != partition) {
+continue;
+ }
+ outputStream.reset();
+
+ datumWriter.write(genericRecord, binaryEncoder);
+ binaryEncoder.flush();
+
+ byte[] bytes = outputStream.toByteArray();
+ // contiguous offsets
+ messageOffsets.add(offset++);
+ messageBytes.add(bytes);
+}
+ }
+} catch (Exception e) {
+ LOGGER.error("Could not create {}",
FakePartitionLevelConsumer.class.getName(), e);
+} finally {
+ FileUtils.deleteQuietly(outputDir);
+}
+ }
+
+ @Override
+ public MessageBatch fetchMessages(long startOffset, long endOffset, int
timeoutMillis) throws TimeoutException {
+if (startOffset >= FakeStreamConfigUtils.getLargestOffset()) {
Review comment:
Instead of unpacking to a tempdir, is it possible to just include the loop
in this place, and get the messages from the original avro file? If this does
not seem straight-forward, you can add that as a TODO and work on it later.
Just trying to make tests go faster, and try not to fill out tempdir.
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.
For queries about this service,
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #4233: Remove depependency on Kafka stream implementation classes from pinot-core and pinot-controller tests
mcvsubbu commented on a change in pull request #4233: Remove depependency on
Kafka stream implementation classes from pinot-core and pinot-controller tests
URL: https://github.com/apache/incubator-pinot/pull/4233#discussion_r287474937
##
File path:
pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionLevelConsumer.java
##
@@ -0,0 +1,121 @@
+/**
+ * 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.core.realtime.impl.fakestream;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.realtime.stream.MessageBatch;
+import org.apache.pinot.core.realtime.stream.PartitionLevelConsumer;
+import org.apache.pinot.core.realtime.stream.StreamConfig;
+import org.apache.pinot.core.util.AvroUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils.*;
+
+
+/**
+ * Implementation of {@link PartitionLevelConsumer} for fake stream
+ * Unpacks tar files in
/resources/data/On_Time_Performance_2014_partition_.tar.gz as source
of messages
Review comment:
Good to add a comment saying that we unpack the tar files into tempdir. Not
sure if we will fill up tempdir
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] kishoreg edited a comment on issue #3723: Remove auto-gen class files that were checked-in
kishoreg edited a comment on issue #3723: Remove auto-gen class files that were checked-in URL: https://github.com/apache/incubator-pinot/pull/3723#issuecomment-495746485 There are a couple of things we need to consider here 1. we will have to modify Travis script to install thrift before triggering the build (this is not hard, we can use thrift docker images) 2. Adds the additional step of installing thrift ( one-time) in order to build Pinot locally. We will have to update the build instructions. The only way to avoid the above steps is to migrate to Avro or Protobuf but that will be a bigger change. If there are no concerns, we can add the steps to install thrift to Travis script. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] kishoreg commented on issue #3723: Remove auto-gen class files that were checked-in
kishoreg commented on issue #3723: Remove auto-gen class files that were checked-in URL: https://github.com/apache/incubator-pinot/pull/3723#issuecomment-495746485 There are a couple of things we need to consider here 1. we will have to modify Travis script to install thrift before triggering the build (this is not hard, we can use thrift docker images) 2. Adds the additional step of installing thrift in order to build Pinot locally. Currently, since the generated code is checked. This is a one-time thing The only way to avoid the above steps is to migrate to Avro or Protobuf but that will be a bigger change. If there are no concerns, we can add the steps to install thrift to Travis script. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on issue #3723: Remove auto-gen class files that were checked-in
snleee commented on issue #3723: Remove auto-gen class files that were checked-in URL: https://github.com/apache/incubator-pinot/pull/3723#issuecomment-495733441 > This is great! @snleee do we need to add extra license for the new dependency? @Jackie-Jiang `thrift` is an apache project so there is no issue. We will clean up `LICENSE` when we do the next release. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on issue #4216: PQL -> SQL enhancement - phase 1 - new Pinot Query Struct
Jackie-Jiang commented on issue #4216: PQL -> SQL enhancement - phase 1 - new Pinot Query Struct URL: https://github.com/apache/incubator-pinot/pull/4216#issuecomment-495729649 @fx19880617 Yes, that is definitely better 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #3723: Remove auto-gen class files that were checked-in
Jackie-Jiang commented on a change in pull request #3723: Remove auto-gen class
files that were checked-in
URL: https://github.com/apache/incubator-pinot/pull/3723#discussion_r287458225
##
File path:
pinot-common/src/main/java/org/apache/pinot/common/config/Deserializer.java
##
@@ -398,8 +398,9 @@ public ConfigIncluder withFallback(ConfigIncluder
fallback) {
config = config.resolve();
try {
- return deserialize(clazz,
io.vavr.collection.HashSet.ofAll(config.entrySet())
- .toMap(entry -> Tuple.of(entry.getKey(),
entry.getValue().unwrapped())), "");
+ Map map =
io.vavr.collection.HashSet.ofAll(config.entrySet())
Review comment:
Anything changed here?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #3723: Remove auto-gen class files that were checked-in
Jackie-Jiang commented on a change in pull request #3723: Remove auto-gen class files that were checked-in URL: https://github.com/apache/incubator-pinot/pull/3723#discussion_r287458370 ## File path: pinot-common/pom.xml ## @@ -72,6 +72,25 @@ org.apache.maven.plugins maven-enforcer-plugin + Review comment: Can you reformat this file? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] kishoreg commented on issue #4219: Support SQL in Pinot.
kishoreg commented on issue #4219: Support SQL in Pinot. URL: https://github.com/apache/incubator-pinot/issues/4219#issuecomment-495722312 @walterddr I looked at the Jira link. If my understanding is right, I think the metadata it refers to is global and not per segment. In the case of Pinot, the physical plan is optimized on a per segment basis based on its metadata (cardinality, indexing, star-tree, dictionary etc) Having said that, in future if we start supporting nested queries and look up joins, we might consider using the optimizers that come with calcite. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[GitHub] [incubator-pinot] jackjlli merged pull request #4236: Add metric on ControllerLeadershipManager
jackjlli merged pull request #4236: Add metric on ControllerLeadershipManager URL: https://github.com/apache/incubator-pinot/pull/4236 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
[incubator-pinot] branch master updated: Add metric on ControllerLeadershipManager (#4236)
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 45a8430 Add metric on ControllerLeadershipManager (#4236)
45a8430 is described below
commit 45a8430bc3d981b44e2d8e242c75480d7dc593e6
Author: Jialiang Li
AuthorDate: Fri May 24 08:53:43 2019 -0700
Add metric on ControllerLeadershipManager (#4236)
---
.../org/apache/pinot/common/metrics/ControllerGauge.java | 3 +++
.../pinot/controller/ControllerLeadershipManager.java| 13 -
.../org/apache/pinot/controller/ControllerStarter.java | 5 +++--
.../realtime/PinotLLCRealtimeSegmentManagerTest.java | 9 +++--
.../helix/core/realtime/SegmentCompletionTest.java | 16 +++-
5 files changed, 36 insertions(+), 10 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
index 052979a..97fc4b6 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
@@ -45,6 +45,9 @@ public enum ControllerGauge implements AbstractMetrics.Gauge {
DISABLED_TABLE_COUNT("TableCount", true),
PERIODIC_TASK_NUM_TABLES_PROCESSED("PeriodicTaskNumTablesProcessed", true),
+ // Pinot controller leader
+ PINOT_CONTROLLER_LEADER("PinotControllerLeader", true),
+
// Number of extra live instances needed
SHORT_OF_LIVE_INSTANCES("ShortOfLiveInstances", false),
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerLeadershipManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerLeadershipManager.java
index 31eef72..035b22d 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerLeadershipManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerLeadershipManager.java
@@ -22,6 +22,8 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.helix.HelixManager;
import org.apache.helix.api.listeners.ControllerChangeListener;
+import org.apache.pinot.common.metrics.ControllerGauge;
+import org.apache.pinot.common.metrics.ControllerMetrics;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,13 +37,16 @@ public class ControllerLeadershipManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(ControllerLeadershipManager.class);
private HelixManager _helixManager;
+ private ControllerMetrics _controllerMetrics;
private volatile boolean _amILeader = false;
private Map _subscribers = new
ConcurrentHashMap<>();
- public ControllerLeadershipManager(HelixManager helixManager) {
+ public ControllerLeadershipManager(HelixManager helixManager,
ControllerMetrics controllerMetrics) {
_helixManager = helixManager;
+_controllerMetrics = controllerMetrics;
_helixManager.addControllerListener((ControllerChangeListener)
notificationContext -> onControllerChange());
+
_controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER,
0L);
}
/**
@@ -81,11 +86,17 @@ public class ControllerLeadershipManager {
}
private void onBecomingLeader() {
+long startTimeMs = System.currentTimeMillis();
+
_controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER,
1L);
_subscribers.forEach((k, v) -> v.onBecomingLeader());
+LOGGER.info("Finished on becoming leader in {}ms",
(System.currentTimeMillis() - startTimeMs));
}
private void onBecomingNonLeader() {
+long startTimeMs = System.currentTimeMillis();
+
_controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER,
0L);
_subscribers.forEach((k, v) -> v.onBecomingNonLeader());
+LOGGER.info("Finished on becoming non-leader in {}ms",
(System.currentTimeMillis() - startTimeMs));
}
/**
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
index fd3169c..7e09e7a 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
@@ -224,6 +224,7 @@ public class ControllerStarter {
// Emit helix controller metrics
_controllerMetrics.addCallbackGauge(CommonConstants.Helix.INSTANCE_CONNECTED_METRIC_NAME,
() -> _helixControllerManager.isConnected() ? 1L : 0L);
+// Deprecated, since getting the leadership of Helix does not mean Helix
has been ready for pinot.
_controllerMetrics.addCallbackGauge("helix.leader", () ->
_helixControllerManager.isLeader() ? 1L
[GitHub] [incubator-pinot] codecov-io edited a comment on issue #4216: PQL -> SQL enhancement - phase 1 - new Pinot Query Struct
codecov-io edited a comment on issue #4216: PQL -> SQL enhancement - phase 1 - new Pinot Query Struct URL: https://github.com/apache/incubator-pinot/pull/4216#issuecomment-493376612 # [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/4216?src=pr=h1) Report > Merging [#4216](https://codecov.io/gh/apache/incubator-pinot/pull/4216?src=pr=desc) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/7bb797d1ba1a8e2c1f1df33fb6b847e154df6423?src=pr=desc) will **decrease** coverage by `1.95%`. > The diff coverage is `33.33%`. [](https://codecov.io/gh/apache/incubator-pinot/pull/4216?src=pr=tree) ```diff @@ Coverage Diff @@ ## master#4216 +/- ## - Coverage 67.39% 65.44% -1.96% Complexity 20 20 Files 1041 1051 +10 Lines 5150554676+3171 Branches 7216 7807 +591 + Hits 3471235782+1070 - Misses1442816297+1869 - Partials 2365 2597 +232 ``` | [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/4216?src=pr=tree) | Coverage Δ | Complexity Δ | | |---|---|---|---| | [...e/pinot/pql/parsers/pql2/ast/PredicateAstNode.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9wcWwvcGFyc2Vycy9wcWwyL2FzdC9QcmVkaWNhdGVBc3ROb2RlLmphdmE=) | `50% <ø> (ø)` | `0 <0> (ø)` | :arrow_down: | | [...rg/apache/pinot/common/request/FilterOperator.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vcmVxdWVzdC9GaWx0ZXJPcGVyYXRvci5qYXZh) | `91.3% <ø> (ø)` | `0 <0> (ø)` | :arrow_down: | | [...t/pql/parsers/pql2/ast/BooleanOperatorAstNode.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9wcWwvcGFyc2Vycy9wcWwyL2FzdC9Cb29sZWFuT3BlcmF0b3JBc3ROb2RlLmphdmE=) | `27.77% <0%> (-3.48%)` | `0 <0> (ø)` | | | [...apache/pinot/common/request/InstanceRequestV2.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vcmVxdWVzdC9JbnN0YW5jZVJlcXVlc3RWMi5qYXZh) | `0% <0%> (ø)` | `0 <0> (?)` | | | [...not/pql/parsers/pql2/ast/PredicateListAstNode.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9wcWwvcGFyc2Vycy9wcWwyL2FzdC9QcmVkaWNhdGVMaXN0QXN0Tm9kZS5qYXZh) | `93.02% <100%> (+3.93%)` | `0 <0> (ø)` | :arrow_down: | | [...che/pinot/pql/parsers/pql2/ast/OptionsAstNode.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9wcWwvcGFyc2Vycy9wcWwyL2FzdC9PcHRpb25zQXN0Tm9kZS5qYXZh) | `100% <100%> (ø)` | `0 <0> (ø)` | :arrow_down: | | [...ache/pinot/pql/parsers/pql2/ast/SelectAstNode.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9wcWwvcGFyc2Vycy9wcWwyL2FzdC9TZWxlY3RBc3ROb2RlLmphdmE=) | `96.34% <100%> (+0.68%)` | `0 <0> (ø)` | :arrow_down: | | [...ache/pinot/pql/parsers/pql2/ast/HavingAstNode.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9wcWwvcGFyc2Vycy9wcWwyL2FzdC9IYXZpbmdBc3ROb2RlLmphdmE=) | `100% <100%> (ø)` | `0 <0> (ø)` | :arrow_down: | | [.../pql/parsers/pql2/ast/OutputColumnListAstNode.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9wcWwvcGFyc2Vycy9wcWwyL2FzdC9PdXRwdXRDb2x1bW5MaXN0QXN0Tm9kZS5qYXZh) | `70% <100%> (+7.5%)` | `0 <0> (ø)` | :arrow_down: | | [...pache/pinot/common/utils/request/RequestUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdXRpbHMvcmVxdWVzdC9SZXF1ZXN0VXRpbHMuamF2YQ==) | `96.58% <100%> (+0.58%)` | `0 <0> (ø)` | :arrow_down: | | ... and [78 more](https://codecov.io/gh/apache/incubator-pinot/pull/4216/diff?src=pr=tree-more) | | -- [Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/4216?src=pr=continue). > **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta) > `Δ = absolute (impact)`,
