This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 46ff7489c59 Remove deprecated Kafka high-level-consumer (HLC) code
(#18931)
46ff7489c59 is described below
commit 46ff7489c59a6559d3ccca2d732d2d105317ea1f
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Jul 8 13:03:49 2026 -0700
Remove deprecated Kafka high-level-consumer (HLC) code (#18931)
---
.../config/feature-test-2-realtime.json | 1 -
.../config/feature-test-2-realtime.json | 1 -
.../config/feature-test-3-realtime.json | 1 -
.../airlineStats_realtime_table_config.json | 2 -
helm/pinot/pinot-github-events-setup.yml | 1 -
helm/pinot/pinot-realtime-quickstart.yml | 4 -
.../pinot/common/metadata/ZKMetadataProvider.java | 19 ---
.../metadata/instance/InstanceZKMetadata.java | 158 ---------------------
.../common/metadata/InstanceZKMetadataTest.java | 71 ---------
.../metadata/MetadataEqualsHashCodeTest.java | 3 -
.../realtime/PinotLLCRealtimeSegmentManager.java | 13 +-
.../helix/core/relocation/SegmentRelocator.java | 13 +-
.../pinot/controller/helix/ControllerTest.java | 1 -
.../RealtimeOffsetAutoResetManagerTest.java | 4 -
...PartialUpsertTableRebalanceIntegrationTest.java | 10 +-
.../stream/kafka/KafkaStreamConfigProperties.java | 12 --
.../pinot/plugin/stream/pulsar/PulsarConfig.java | 13 +-
.../PulsarPartitionLevelConnectionHandler.java | 2 +-
.../plugin/stream/pulsar/PulsarConfigTest.java | 12 +-
.../indexsegment/mutable/MutableSegmentImpl.java | 27 ----
.../segment/local/segment/creator/Fixtures.java | 2 -
.../pinot/segment/local/utils/IngestionUtils.java | 14 --
.../pinot/spi/stream/PartitionLevelConsumer.java | 70 ---------
.../spi/stream/PartitionLevelStreamConfig.java | 30 ----
.../pinot/spi/stream/StreamConfigProperties.java | 13 +-
.../pinot/spi/stream/StreamConsumerFactory.java | 31 ++--
.../pinot/spi/stream/StreamLevelConsumer.java | 62 --------
.../pinot/tools/utils/KafkaStarterUtils.java | 2 -
.../githubEvents_realtime_table_config.json | 1 -
.../airlineStats_realtime_table_config.json | 1 -
.../docker/airlineStats_realtime_table_config.json | 2 -
.../dailySales_realtime_table_config.json | 1 -
.../fineFoodReviews_realtime_table_config.json | 1 -
...neFoodReviews_part_0_realtime_table_config.json | 1 -
...neFoodReviews_part_1_realtime_table_config.json | 1 -
.../githubEvents_realtime_table_config.json | 1 -
.../meetupRsvp_realtime_table_config.json | 1 -
...eetupRsvpComplexType_realtime_table_config.json | 1 -
.../meetupRsvpJson_realtime_table_config.json | 1 -
...lRequestMergedEvents_realtime_table_config.json | 1 -
...lRequestMergedEvents_realtime_table_config.json | 1 -
...upsertJsonMeetupRsvp_realtime_table_config.json | 1 -
.../upsertMeetupRsvp_realtime_table_config.json | 1 -
...ertPartialMeetupRsvp_realtime_table_config.json | 1 -
44 files changed, 38 insertions(+), 571 deletions(-)
diff --git
a/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json
b/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json
index 1a3980d19dc..ab05a2710b7 100644
---
a/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json
+++
b/compatibility-verifier/multi-stage-query-engine-test-suite/config/feature-test-2-realtime.json
@@ -45,7 +45,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.broker.list": "localhost:19092",
- "stream.kafka.zk.broker.url": "localhost:2181/kafka",
"stream.kafka.consumer.prop.auto.offset.reset": "largest"
}
},
diff --git
a/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json
b/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json
index 1a3980d19dc..ab05a2710b7 100644
---
a/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json
+++
b/compatibility-verifier/sample-test-suite/config/feature-test-2-realtime.json
@@ -45,7 +45,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.broker.list": "localhost:19092",
- "stream.kafka.zk.broker.url": "localhost:2181/kafka",
"stream.kafka.consumer.prop.auto.offset.reset": "largest"
}
},
diff --git
a/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json
b/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json
index 59a59116ef9..09a5f73ed6d 100644
---
a/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json
+++
b/compatibility-verifier/sample-test-suite/config/feature-test-3-realtime.json
@@ -28,7 +28,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.broker.list": "localhost:19092",
- "stream.kafka.zk.broker.url": "localhost:2181/kafka",
"stream.kafka.consumer.prop.auto.offset.reset": "largest"
}
},
diff --git
a/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json
b/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json
index d4177f60d01..25794fbb598 100644
---
a/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json
+++
b/docker/images/pinot/examples/docker/table-configs/airlineStats_realtime_table_config.json
@@ -18,8 +18,6 @@
"stream.kafka.topic.name": "flights-realtime",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.hlc.zk.connect.string": "pinot-zookeeper:2181/kafka",
- "stream.kafka.zk.broker.url": "pinot-zookeeper:2181/kafka",
"stream.kafka.broker.list": "kafka:9092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000",
diff --git a/helm/pinot/pinot-github-events-setup.yml
b/helm/pinot/pinot-github-events-setup.yml
index 894ab0e9b33..8db97ad1d82 100644
--- a/helm/pinot/pinot-github-events-setup.yml
+++ b/helm/pinot/pinot-github-events-setup.yml
@@ -47,7 +47,6 @@ data:
"stream.kafka.topic.name": "pullRequestMergedEvents",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.zk.broker.url": "kafka-zookeeper:2181",
"stream.kafka.broker.list": "kafka:9092",
"realtime.segment.flush.threshold.time": "12h",
"realtime.segment.flush.threshold.size": "100000",
diff --git a/helm/pinot/pinot-realtime-quickstart.yml
b/helm/pinot/pinot-realtime-quickstart.yml
index 5b2e24bfd29..f537d67cb6c 100644
--- a/helm/pinot/pinot-realtime-quickstart.yml
+++ b/helm/pinot/pinot-realtime-quickstart.yml
@@ -44,8 +44,6 @@ data:
"stream.kafka.topic.name": "flights-realtime",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.hlc.zk.connect.string": "kafka-zookeeper:2181",
- "stream.kafka.zk.broker.url": "kafka-zookeeper:2181",
"stream.kafka.broker.list": "kafka:9092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000",
@@ -79,8 +77,6 @@ data:
"stream.kafka.decoder.class.name":
"org.apache.pinot.core.realtime.stream.SimpleAvroMessageDecoder",
"stream.kafka.decoder.prop.schema":
"{\"type\":\"record\",\"name\":\"Flight\",\"namespace\":\"pinot\",\"fields\":[{\"name\":\"DaysSinceEpoch\",\"type\":[\"int\"]},{\"name\":\"Year\",\"type\":[\"int\"]},{\"name\":\"Quarter\",\"type\":[\"int\"]},{\"name\":\"Month\",\"type\":[\"int\"]},{\"name\":\"DayofMonth\",\"type\":[\"int\"]},{\"name\":\"DayOfWeek\",\"type\":[\"int\"]},{\"name\":\"FlightDate\",\"type\":[\"string\"]},{\"name\":\"UniqueCarrier\",\"type\":[\"string\"]},{\"name\":
[...]
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.hlc.zk.connect.string": "kafka-zookeeper:2181",
- "stream.kafka.zk.broker.url": "kafka-zookeeper:2181",
"stream.kafka.broker.list": "kafka:9092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000",
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
index a0f98058739..b3154724622 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
@@ -36,7 +36,6 @@ import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.zkclient.exception.ZkBadVersionException;
import org.apache.pinot.common.assignment.InstancePartitions;
-import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.LogicalTableConfigUtils;
@@ -79,7 +78,6 @@ public class ZKMetadataProvider {
private static final String PROPERTYSTORE_DATABASE_CONFIGS_PREFIX =
"/CONFIGS/DATABASE";
private static final String PROPERTYSTORE_TABLE_CONFIGS_PREFIX =
"/CONFIGS/TABLE";
private static final String PROPERTYSTORE_USER_CONFIGS_PREFIX =
"/CONFIGS/USER";
- private static final String PROPERTYSTORE_INSTANCE_CONFIGS_PREFIX =
"/CONFIGS/INSTANCE";
private static final String PROPERTYSTORE_CLUSTER_CONFIGS_PREFIX =
"/CONFIGS/CLUSTER";
private static final String PROPERTYSTORE_SEGMENT_LINEAGE =
"/SEGMENT_LINEAGE";
private static final String PROPERTYSTORE_MINION_TASK_METADATA_PREFIX =
"/MINION_TASK_METADATA";
@@ -238,23 +236,6 @@ public class ZKMetadataProvider {
setTableConfig(propertyStore, offlineTableName, znRecord);
}
- public static void setInstanceZKMetadata(ZkHelixPropertyStore<ZNRecord>
propertyStore,
- InstanceZKMetadata instanceZKMetadata) {
- ZNRecord znRecord = instanceZKMetadata.toZNRecord();
- propertyStore.set(StringUtil.join("/",
PROPERTYSTORE_INSTANCE_CONFIGS_PREFIX, instanceZKMetadata.getId()), znRecord,
- AccessOption.PERSISTENT);
- }
-
- public static InstanceZKMetadata
getInstanceZKMetadata(ZkHelixPropertyStore<ZNRecord> propertyStore,
- String instanceId) {
- ZNRecord znRecord = propertyStore.get(StringUtil.join("/",
PROPERTYSTORE_INSTANCE_CONFIGS_PREFIX, instanceId), null,
- AccessOption.PERSISTENT);
- if (znRecord == null) {
- return null;
- }
- return new InstanceZKMetadata(znRecord);
- }
-
public static String constructPropertyStorePathForSegment(String
resourceName, String segmentName) {
return StringUtil.join("/", PROPERTYSTORE_SEGMENTS_PREFIX, resourceName,
segmentName);
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metadata/instance/InstanceZKMetadata.java
b/pinot-common/src/main/java/org/apache/pinot/common/metadata/instance/InstanceZKMetadata.java
deleted file mode 100644
index d93d51f8a05..00000000000
---
a/pinot-common/src/main/java/org/apache/pinot/common/metadata/instance/InstanceZKMetadata.java
+++ /dev/null
@@ -1,158 +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.common.metadata.instance;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
-import org.apache.pinot.common.metadata.ZKMetadata;
-import org.apache.pinot.spi.utils.StringUtil;
-import org.apache.pinot.spi.utils.builder.TableNameBuilder;
-
-import static org.apache.pinot.spi.utils.EqualityUtils.hashCodeOf;
-import static org.apache.pinot.spi.utils.EqualityUtils.isEqual;
-import static org.apache.pinot.spi.utils.EqualityUtils.isNullOrNotSameClass;
-import static org.apache.pinot.spi.utils.EqualityUtils.isSameReference;
-
-
-public final class InstanceZKMetadata implements ZKMetadata {
- private static final String KAFKA_HIGH_LEVEL_CONSUMER_GROUP_MAP =
"KAFKA_HLC_GROUP_MAP";
- private static final String KAFKA_HIGH_LEVEL_CONSUMER_PARTITION_MAP =
"KAFKA_HLC_PARTITION_MAP";
- private String _id = null;
- private String _instanceName = null;
- private int _instancePort;
- private String _instanceType = null;
-
- // Only care about realtime resources for now
- private Map<String, String> _groupIdMap = new HashMap<String, String>();
- private Map<String, String> _partitionMap = new HashMap<String, String>();
-
- public InstanceZKMetadata() {
- }
-
- public InstanceZKMetadata(ZNRecord record) {
- _id = record.getId();
- setInstanceConfigFromId(_id);
-
_groupIdMap.putAll(record.getMapField(KAFKA_HIGH_LEVEL_CONSUMER_GROUP_MAP));
-
_partitionMap.putAll(record.getMapField(KAFKA_HIGH_LEVEL_CONSUMER_PARTITION_MAP));
- }
-
- private void setInstanceConfigFromId(String id) {
- String[] instanceConfigs = id.split("_");
- assert instanceConfigs.length == 3;
- setInstanceType(instanceConfigs[0]);
- setInstanceName(instanceConfigs[1]);
- setInstancePort(Integer.parseInt(instanceConfigs[2]));
- }
-
- public int getInstancePort() {
- return _instancePort;
- }
-
- public void setInstancePort(int instancePort) {
- _instancePort = instancePort;
- }
-
- public String getInstanceType() {
- return _instanceType;
- }
-
- public void setInstanceType(String instanceType) {
- _instanceType = instanceType;
- }
-
- public String getId() {
- if (_id == null) {
- _id = buildIdFromInstanceConfig();
- }
- return _id;
- }
-
- public void setId(String id) {
- _id = id;
- }
-
- public String getInstanceName() {
- return _instanceName;
- }
-
- public void setInstanceName(String instanceName) {
- _instanceName = instanceName;
- }
-
- public String getGroupId(String resourceName) {
- return
_groupIdMap.get(TableNameBuilder.REALTIME.tableNameWithType(resourceName));
- }
-
- public void setGroupId(String resourceName, String groupId) {
- _groupIdMap.put(TableNameBuilder.REALTIME.tableNameWithType(resourceName),
groupId);
- }
-
- public String getPartition(String resourceName) {
- return
_partitionMap.get(TableNameBuilder.REALTIME.tableNameWithType(resourceName));
- }
-
- public void setPartition(String resourceName, String partition) {
-
_partitionMap.put(TableNameBuilder.REALTIME.tableNameWithType(resourceName),
partition);
- }
-
- public void removeResource(String resourceName) {
- _groupIdMap.remove(resourceName);
- _partitionMap.remove(resourceName);
- }
-
- @Override
- public ZNRecord toZNRecord() {
- ZNRecord znRecord = new ZNRecord(getId());
- znRecord.setMapField(KAFKA_HIGH_LEVEL_CONSUMER_GROUP_MAP, _groupIdMap);
- znRecord.setMapField(KAFKA_HIGH_LEVEL_CONSUMER_PARTITION_MAP,
_partitionMap);
- return znRecord;
- }
-
- private String buildIdFromInstanceConfig() {
- return StringUtil.join("_", _instanceType, _instanceName, _instancePort +
"");
- }
-
- @Override
- public boolean equals(Object instanceMetadata) {
- if (isSameReference(this, instanceMetadata)) {
- return true;
- }
-
- if (isNullOrNotSameClass(this, instanceMetadata)) {
- return false;
- }
-
- InstanceZKMetadata metadata = (InstanceZKMetadata) instanceMetadata;
- return isEqual(_id, metadata._id) && isEqual(_instanceName,
metadata._instanceName) && isEqual(_instanceType,
- metadata._instanceType) && isEqual(_instancePort,
metadata._instancePort) && isEqual(_groupIdMap,
- metadata._groupIdMap) && isEqual(_partitionMap,
metadata._partitionMap);
- }
-
- @Override
- public int hashCode() {
- int result = hashCodeOf(_id);
- result = hashCodeOf(result, _instanceName);
- result = hashCodeOf(result, _instancePort);
- result = hashCodeOf(result, _instanceType);
- result = hashCodeOf(result, _groupIdMap);
- result = hashCodeOf(result, _partitionMap);
- return result;
- }
-}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/metadata/InstanceZKMetadataTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/metadata/InstanceZKMetadataTest.java
deleted file mode 100644
index 027d8c5d409..00000000000
---
a/pinot-common/src/test/java/org/apache/pinot/common/metadata/InstanceZKMetadataTest.java
+++ /dev/null
@@ -1,71 +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.common.metadata;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
-import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
-
-public class InstanceZKMetadataTest {
-
- @Test
- public void instanceZKMetadataConversionTest() {
- ZNRecord znRecord = getTestInstanceZNRecord();
- InstanceZKMetadata instanceMetadataFromZNRecord = new
InstanceZKMetadata(znRecord);
-
- InstanceZKMetadata instanceMetadata = getTestInstanceMetadata();
- ZNRecord znRecordFromMetadata = instanceMetadata.toZNRecord();
-
- Assert.assertTrue(MetadataUtils.comparisonZNRecords(znRecord,
znRecordFromMetadata));
- Assert.assertTrue(instanceMetadata.equals(instanceMetadataFromZNRecord));
-
- Assert.assertTrue(instanceMetadata.equals(new
InstanceZKMetadata(instanceMetadata.toZNRecord())));
- Assert.assertTrue(MetadataUtils.comparisonZNRecords(znRecord, new
InstanceZKMetadata(znRecord).toZNRecord()));
- }
-
- private ZNRecord getTestInstanceZNRecord() {
- ZNRecord record = new ZNRecord("Server_localhost_1234");
- Map<String, String> groupIdMap = new HashMap<>();
- Map<String, String> partitionMap = new HashMap<>();
-
- for (int i = 0; i < 10; i++) {
- groupIdMap.put("testRes" + i + "_REALTIME", "groupId" + i);
- partitionMap.put("testRes" + i + "_REALTIME", "part" + i);
- }
- record.setMapField("KAFKA_HLC_GROUP_MAP", groupIdMap);
- record.setMapField("KAFKA_HLC_PARTITION_MAP", partitionMap);
- return record;
- }
-
- private InstanceZKMetadata getTestInstanceMetadata() {
- InstanceZKMetadata instanceMetadata = new InstanceZKMetadata();
- instanceMetadata.setInstanceType("Server");
- instanceMetadata.setInstanceName("localhost");
- instanceMetadata.setInstancePort(1234);
- for (int i = 0; i < 10; i++) {
- instanceMetadata.setGroupId("testRes" + i, "groupId" + i);
- instanceMetadata.setPartition("testRes" + i, "part" + i);
- }
- return instanceMetadata;
- }
-}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/metadata/MetadataEqualsHashCodeTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/metadata/MetadataEqualsHashCodeTest.java
index b2a6236337e..96ec221e487 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/metadata/MetadataEqualsHashCodeTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/metadata/MetadataEqualsHashCodeTest.java
@@ -20,7 +20,6 @@ package org.apache.pinot.common.metadata;
import nl.jqno.equalsverifier.EqualsVerifier;
import nl.jqno.equalsverifier.Warning;
-import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.MetricFieldSpec;
@@ -37,8 +36,6 @@ public class MetadataEqualsHashCodeTest {
@Test
public void testEqualsAndHashCode() {
-
EqualsVerifier.forClass(InstanceZKMetadata.class).suppress(Warning.NULL_FIELDS,
Warning.NONFINAL_FIELDS)
- .usingGetClass().verify();
EqualsVerifier.forClass(DimensionFieldSpec.class).suppress(Warning.NULL_FIELDS,
Warning.NONFINAL_FIELDS)
.usingGetClass().verify();
EqualsVerifier.forClass(MetricFieldSpec.class).suppress(Warning.NULL_FIELDS,
Warning.NONFINAL_FIELDS)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index 9bc0d4e8a70..d1d1f3bb306 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -379,10 +379,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
}
}
- /**
- * Sets up the initial segments for a new LLC real-time table.
- * <p>NOTE: the passed in IdealState may contain HLC segments if both HLC
and LLC are configured.
- */
+ /// Sets up the initial segments for a new real-time table.
public void setUpNewTable(TableConfig tableConfig, IdealState idealState) {
List<StreamConfig> streamConfigs =
IngestionConfigUtils.getStreamConfigs(tableConfig);
List<StreamMetadata> streamMetadataList =
@@ -390,12 +387,8 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
setUpNewTable(tableConfig, idealState, streamMetadataList);
}
- /**
- * Sets up the initial segments for a new LLC real-time table.
- * <p>NOTE: the passed in IdealState may contain HLC segments if both HLC
and LLC are configured.
- */
- public void setUpNewTable(TableConfig tableConfig, IdealState idealState,
- List<StreamMetadata> streamMetadataList) {
+ /// Sets up the initial segments for a new real-time table.
+ public void setUpNewTable(TableConfig tableConfig, IdealState idealState,
List<StreamMetadata> streamMetadataList) {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
index 7c55131ca5f..644d857915b 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
@@ -55,14 +55,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- * Periodic task to run rebalancer in background to:
- * <ol>
- * <li> Relocate COMPLETED segments to tag overrides
- * <li> Relocate ONLINE segments to tiers if tier configs are set
- * </ol>
- * Allow at most one replica unavailable during rebalance. Not applicable for
HLC tables.
- */
+/// Periodic task to run rebalancer in background to:
+/// - Relocate COMPLETED segments to tag overrides
+/// - Relocate ONLINE segments to tiers if tier configs are set
+///
+/// Allow at most one replica unavailable during rebalance.
public class SegmentRelocator extends ControllerPeriodicTask<Void> {
private static final Logger LOGGER =
LoggerFactory.getLogger(SegmentRelocator.class);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
index af425d23ef2..e6d73b2ebb0 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
@@ -132,7 +132,6 @@ public class ControllerTest {
// Default ControllerTest instance settings
public static final int DEFAULT_MIN_NUM_REPLICAS = 2;
public static final int DEFAULT_NUM_BROKER_INSTANCES = 3;
- // NOTE: To add HLC realtime table, number of Server instances must be
multiple of replicas
public static final int DEFAULT_NUM_SERVER_INSTANCES = 4;
public static final int DEFAULT_NUM_MINION_INSTANCES = 2;
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/RealtimeOffsetAutoResetManagerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/RealtimeOffsetAutoResetManagerTest.java
index fd2084a2e3e..4ce8d8be80b 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/RealtimeOffsetAutoResetManagerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/RealtimeOffsetAutoResetManagerTest.java
@@ -257,7 +257,6 @@ public class RealtimeOffsetAutoResetManagerTest {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put("streamType", "kafka");
streamConfigMap.put("stream.kafka.topic.name", TOPIC_NAME);
- streamConfigMap.put("stream.kafka.consumer.type", "simple");
streamConfigMap.put("realtime.segment.offsetAutoReset.timeSecThreshold",
"1800");
streamConfigMap.put("stream.kafka.decoder.class.name", "testDecoder");
StreamIngestionConfig streamIngestionConfig = new
StreamIngestionConfig(List.of(streamConfigMap));
@@ -274,7 +273,6 @@ public class RealtimeOffsetAutoResetManagerTest {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put("streamType", "kafka");
streamConfigMap.put("stream.kafka.topic.name", TOPIC_NAME);
- streamConfigMap.put("stream.kafka.consumer.type", "simple");
streamConfigMap.put("realtime.segment.offsetAutoReset.timeSecThreshold",
"1800");
streamConfigMap.put("stream.kafka.decoder.class.name", "testDecoder");
StreamIngestionConfig streamIngestionConfig = new
StreamIngestionConfig(List.of(streamConfigMap));
@@ -292,7 +290,6 @@ public class RealtimeOffsetAutoResetManagerTest {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put("streamType", "kafka");
streamConfigMap.put("stream.kafka.topic.name", TOPIC_NAME);
- streamConfigMap.put("stream.kafka.consumer.type", "simple");
streamConfigMap.put("realtime.segment.offsetAutoReset.timeSecThreshold",
"1800");
streamConfigMap.put("stream.kafka.decoder.class.name", "testDecoder");
@@ -326,7 +323,6 @@ public class RealtimeOffsetAutoResetManagerTest {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put("streamType", "kafka");
streamConfigMap.put("stream.kafka.topic.name", TOPIC_NAME);
- streamConfigMap.put("stream.kafka.consumer.type", "simple");
streamConfigMap.put("realtime.segment.offsetAutoReset.timeSecThreshold",
"1800");
streamConfigMap.put("stream.kafka.decoder.class.name", "testDecoder");
StreamIngestionConfig streamIngestionConfig = new
StreamIngestionConfig(List.of(streamConfigMap));
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PartialUpsertTableRebalanceIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PartialUpsertTableRebalanceIntegrationTest.java
index d636a61351f..74b4910b77a 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PartialUpsertTableRebalanceIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PartialUpsertTableRebalanceIntegrationTest.java
@@ -55,6 +55,7 @@ import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+
public class PartialUpsertTableRebalanceIntegrationTest extends
BaseClusterIntegrationTest {
private static final int NUM_SERVERS = 1;
private static final String PRIMARY_KEY_COL = "clientId";
@@ -281,9 +282,8 @@ public class PartialUpsertTableRebalanceIntegrationTest
extends BaseClusterInteg
int maxSequenceNumber = 0;
for (Map.Entry<String, Map<String, String>> entry :
segmentAssignment.entrySet()) {
- String segmentName = entry.getKey();
- if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(entry.getKey());
+ if (llcSegmentName != null) {
maxSequenceNumber = Math.max(maxSequenceNumber,
llcSegmentName.getSequenceNumber());
}
}
@@ -298,8 +298,8 @@ public class PartialUpsertTableRebalanceIntegrationTest
extends BaseClusterInteg
assertEquals(instanceStateMap.size(), 1);
Map.Entry<String, String> instanceIdAndState =
instanceStateMap.entrySet().iterator().next();
String state = instanceIdAndState.getValue();
- if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null) {
if (llcSegmentName.getSequenceNumber() < maxSequenceNumber) {
assertEquals(state,
CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE);
} else {
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java
index 9614f436fbf..ee0b6276a30 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java
@@ -41,18 +41,6 @@ public class KafkaStreamConfigProperties {
return Joiner.on(DOT_SEPARATOR).join(StreamConfigProperties.STREAM_PREFIX,
property);
}
- public static class HighLevelConsumer {
- public static final String KAFKA_HLC_BOOTSTRAP_SERVER =
"kafka.hlc.bootstrap.server";
- public static final String KAFKA_HLC_ZK_CONNECTION_STRING =
"kafka.hlc.zk.connect.string";
- public static final String ZK_SESSION_TIMEOUT_MS =
"zookeeper.session.timeout.ms";
- public static final String ZK_CONNECTION_TIMEOUT_MS =
"zookeeper.connection.timeout.ms";
- public static final String ZK_SYNC_TIME_MS = "zookeeper.sync.time.ms";
- public static final String REBALANCE_MAX_RETRIES = "rebalance.max.retries";
- public static final String REBALANCE_BACKOFF_MS = "rebalance.backoff.ms";
- public static final String AUTO_COMMIT_ENABLE = "auto.commit.enable";
- public static final String AUTO_OFFSET_RESET = "auto.offset.reset";
- }
-
public static class LowLevelConsumer {
public static final String KAFKA_BROKER_LIST = "kafka.broker.list";
public static final String KAFKA_BUFFER_SIZE = "kafka.buffer.size";
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfig.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfig.java
index 48e960589ce..675fc37a57f 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfig.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfig.java
@@ -34,10 +34,7 @@ import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
-/**
- * Pulsar specific stream config
- * contains pulsar brokers list, start offset and group id/subscriber id if
using high level consumer/
- */
+/// Pulsar specific stream configs.
public class PulsarConfig {
public static final String STREAM_TYPE = "pulsar";
public static final String BOOTSTRAP_SERVERS = "bootstrap.servers";
@@ -50,7 +47,6 @@ public class PulsarConfig {
public static final String ENABLE_KEY_VALUE_STITCH = "enableKeyValueStitch";
public static final String METADATA_FIELDS = "metadata.fields"; //list of
the metadata fields comma separated
- private final String _subscriberId;
private final String _pulsarTopicName;
private final String _bootstrapServers;
private final String _serviceHttpUrl;
@@ -72,9 +68,8 @@ public class PulsarConfig {
private final boolean _populateMetadata;
private final Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue>
_metadataFields;
- public PulsarConfig(StreamConfig streamConfig, String subscriberId) {
+ public PulsarConfig(StreamConfig streamConfig) {
Map<String, String> streamConfigMap = streamConfig.getStreamConfigsMap();
- _subscriberId = subscriberId;
_pulsarTopicName = streamConfig.getTopicName();
_bootstrapServers = getConfigValue(streamConfigMap, BOOTSTRAP_SERVERS);
@@ -143,10 +138,6 @@ public class PulsarConfig {
return _pulsarTopicName;
}
- public String getSubscriberId() {
- return _subscriberId;
- }
-
public String getBootstrapServers() {
return _bootstrapServers;
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
index 140fab7cba0..60646441d8f 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
@@ -49,7 +49,7 @@ public class PulsarPartitionLevelConnectionHandler implements
Closeable {
* Creates a new instance of {@link PulsarClient} and {@link Reader}
*/
protected PulsarPartitionLevelConnectionHandler(String clientId,
StreamConfig streamConfig) {
- _config = new PulsarConfig(streamConfig, clientId);
+ _config = new PulsarConfig(streamConfig);
_clientId = clientId;
_pulsarClient = createPulsarClient();
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfigTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfigTest.java
index bd4151b878f..6da38ee6f7a 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfigTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConfigTest.java
@@ -59,7 +59,7 @@ public class PulsarConfigTest {
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
PulsarConfig.METADATA_FIELDS),
"messageId,messageIdBytes, publishTime, eventTime, key, topicName, ");
StreamConfig streamConfig = new StreamConfig(TABLE_NAME_WITH_TYPE,
streamConfigMap);
- PulsarConfig pulsarConfig = new PulsarConfig(streamConfig, "testId");
+ PulsarConfig pulsarConfig = new PulsarConfig(streamConfig);
Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue>
metadataFieldsToExtract =
pulsarConfig.getMetadataFields();
Assert.assertEquals(metadataFieldsToExtract.size(), 6);
@@ -78,7 +78,7 @@ public class PulsarConfigTest {
streamConfigMap.put(
StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
StreamConfigProperties.METADATA_POPULATE), "true");
StreamConfig streamConfig = new StreamConfig(TABLE_NAME_WITH_TYPE,
streamConfigMap);
- PulsarConfig pulsarConfig = new PulsarConfig(streamConfig, "testId");
+ PulsarConfig pulsarConfig = new PulsarConfig(streamConfig);
Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue>
metadataFieldsToExtract =
pulsarConfig.getMetadataFields();
Assert.assertEquals(metadataFieldsToExtract.size(), 0);
@@ -90,7 +90,7 @@ public class PulsarConfigTest {
streamConfigMap.put(
StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
StreamConfigProperties.METADATA_POPULATE), "false");
StreamConfig streamConfig = new StreamConfig(TABLE_NAME_WITH_TYPE,
streamConfigMap);
- PulsarConfig pulsarConfig = new PulsarConfig(streamConfig, "testId");
+ PulsarConfig pulsarConfig = new PulsarConfig(streamConfig);
Assert.assertFalse(pulsarConfig.isPopulateMetadata());
Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue>
metadataFieldsToExtract =
pulsarConfig.getMetadataFields();
@@ -105,7 +105,7 @@ public class PulsarConfigTest {
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
PulsarConfig.METADATA_FIELDS),
"messageId,messageIdBytes, publishTime, eventTime, key, topicName, ");
StreamConfig streamConfig = new StreamConfig(TABLE_NAME_WITH_TYPE,
streamConfigMap);
- PulsarConfig pulsarConfig = new PulsarConfig(streamConfig, "testId");
+ PulsarConfig pulsarConfig = new PulsarConfig(streamConfig);
Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue>
metadataFieldsToExtract =
pulsarConfig.getMetadataFields();
Assert.assertFalse(pulsarConfig.isPopulateMetadata());
@@ -128,7 +128,7 @@ public class PulsarConfigTest {
"urn:test:test");
StreamConfig streamConfig = new StreamConfig(TABLE_NAME_WITH_TYPE,
streamConfigMap);
- PulsarConfig pulsarConfig = new PulsarConfig(streamConfig, "testId");
+ PulsarConfig pulsarConfig = new PulsarConfig(streamConfig);
Assert.assertEquals(pulsarConfig.getIssuerUrl(), "http://auth.test.com";);
Assert.assertEquals(pulsarConfig.getCredentialsFilePath(), "file://" +
testFile.toFile().getAbsolutePath());
Assert.assertEquals(pulsarConfig.getAudience(), "urn:test:test");
@@ -149,6 +149,6 @@ public class PulsarConfigTest {
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
PulsarConfig.OAUTH_AUDIENCE),
"urn:test:test");
StreamConfig streamConfig = new StreamConfig(TABLE_NAME_WITH_TYPE,
streamConfigMap);
- new PulsarConfig(streamConfig, "testId");
+ new PulsarConfig(streamConfig);
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
index 0de51c032cb..3eac1210a2a 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
@@ -77,7 +77,6 @@ import org.apache.pinot.segment.local.upsert.UpsertContext;
import org.apache.pinot.segment.local.upsert.UpsertViewManager;
import org.apache.pinot.segment.local.utils.FixedIntArrayOffHeapIdMap;
import org.apache.pinot.segment.local.utils.IdMap;
-import org.apache.pinot.segment.local.utils.IngestionUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.MutableSegment;
import org.apache.pinot.segment.spi.SegmentMetadata;
@@ -149,7 +148,6 @@ public class MutableSegmentImpl implements MutableSegment {
private final String _realtimeTableName;
private final String _segmentName;
private final Schema _schema;
- private final String _timeColumnName;
private final int _capacity;
private final SegmentMetadata _segmentMetadata;
private final boolean _offHeap;
@@ -210,7 +208,6 @@ public class MutableSegmentImpl implements MutableSegment {
_realtimeTableName = config.getTableNameWithType();
_segmentName = config.getSegmentName();
_schema = config.getSchema();
- _timeColumnName = config.getTimeColumnName();
_capacity = config.getCapacity();
SegmentZKMetadata segmentZKMetadata = config.getSegmentZKMetadata();
_segmentMetadata = new
SegmentMetadataImpl(TableNameBuilder.extractRawTableName(_realtimeTableName),
@@ -610,30 +607,6 @@ public class MutableSegmentImpl implements MutableSegment {
}
}
- /**
- * Get min time from the segment, based on the time column, only used by
Kafka HLC.
- */
- @Deprecated
- public long getMinTime() {
- Long minTime =
IngestionUtils.extractTimeValue(_indexContainerMap.get(_timeColumnName)._minValue);
- if (minTime != null) {
- return minTime;
- }
- return Long.MAX_VALUE;
- }
-
- /**
- * Get max time from the segment, based on the time column, only used by
Kafka HLC.
- */
- @Deprecated
- public long getMaxTime() {
- Long maxTime =
IngestionUtils.extractTimeValue(_indexContainerMap.get(_timeColumnName)._maxValue);
- if (maxTime != null) {
- return maxTime;
- }
- return Long.MIN_VALUE;
- }
-
@Override
public boolean index(GenericRow row, @Nullable StreamMessageMetadata
metadata)
throws IOException {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/Fixtures.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/Fixtures.java
index fe53a7f606a..6c6167f62f5 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/Fixtures.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/Fixtures.java
@@ -66,9 +66,7 @@ public class Fixtures {
+ " \"stream.fakeStream.decoder.class.name\":\"%s\","
+ "
\"stream.fakeStream.decoder.prop.schema.registry.rest.url\":\"http://1.2.3.4:1766/schemas\",";
+ "
\"stream.fakeStream.decoder.prop.schema.registry.schema.name\":\"UnknownSchema\","
- + "
\"stream.fakeStream.hlc.zk.connect.string\":\"zoo:2181/kafka-queuing\","
+ " \"stream.fakeStream.topic.name\":\"" + TOPIC_NAME + "\","
- + "
\"stream.fakeStream.zk.broker.url\":\"kafka-broker:2181/kafka-queuing\","
+ " \"streamType\":\"fakeStream\""
+ " }"
+ " },"
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/IngestionUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/IngestionUtils.java
index 92e2193a1fd..c86df21d1f8 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/IngestionUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/IngestionUtils.java
@@ -353,18 +353,4 @@ public final class IngestionUtils {
}
return result;
}
-
- public static Long extractTimeValue(Comparable time) {
- if (time != null) {
- if (time instanceof Number) {
- return ((Number) time).longValue();
- } else {
- String stringValue = time.toString();
- if (StringUtils.isNumeric(stringValue)) {
- return Long.parseLong(stringValue);
- }
- }
- }
- return null;
- }
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelConsumer.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelConsumer.java
deleted file mode 100644
index 73953b2aa42..00000000000
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelConsumer.java
+++ /dev/null
@@ -1,70 +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.spi.stream;
-
-import java.io.Closeable;
-import org.apache.pinot.spi.annotations.InterfaceAudience;
-import org.apache.pinot.spi.annotations.InterfaceStability;
-
-
-/**
- * Interface for a consumer which fetches messages at the partition level of a
stream, for given offsets
- */
-@Deprecated
[email protected]
[email protected]
-public interface PartitionLevelConsumer extends Closeable,
PartitionGroupConsumer {
-
- /**
- * Is here for backward compatibility for a short time.
- * TODO Issue 5359 remove this API once external kafka consumers implements
return of StreamPartitionMsgOffset
- * Fetch messages from the stream between the specified offsets
- * @param startOffset
- * @param endOffset
- * @param timeoutMillis
- * @return
- * @throws java.util.concurrent.TimeoutException
- */
- @Deprecated
- default MessageBatch fetchMessages(long startOffset, long endOffset, int
timeoutMillis)
- throws java.util.concurrent.TimeoutException {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Fetch messages and the per-partition high watermark from Kafka between
the specified offsets.
- *
- * @param startOffset The offset of the first message desired, inclusive
- * @param endOffset The offset of the last message desired, exclusive, or
null
- * @param timeoutMillis Timeout in milliseconds
- * @throws java.util.concurrent.TimeoutException If the operation could not
be completed within {@code timeoutMillis}
- * milliseconds
- * @return An iterable containing messages fetched from the stream partition
and their offsets, as well as the
- * high watermark for this partition.
- */
- @Deprecated
- default MessageBatch fetchMessages(StreamPartitionMsgOffset startOffset,
StreamPartitionMsgOffset endOffset,
- int timeoutMillis)
- throws java.util.concurrent.TimeoutException {
- // TODO Issue 5359 remove this default implementation once all kafka
consumers have migrated to use this API
- long startOffsetLong = ((LongMsgOffset) startOffset).getOffset();
- long endOffsetLong = endOffset == null ? Long.MAX_VALUE : ((LongMsgOffset)
endOffset).getOffset();
- return fetchMessages(startOffsetLong, endOffsetLong, timeoutMillis);
- }
-}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelStreamConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelStreamConfig.java
deleted file mode 100644
index 82092fe3360..00000000000
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelStreamConfig.java
+++ /dev/null
@@ -1,30 +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.spi.stream;
-
-import java.util.Map;
-
-
-@Deprecated
-public class PartitionLevelStreamConfig extends StreamConfig {
-
- public PartitionLevelStreamConfig(String tableNameWithType, Map<String,
String> streamConfigMap) {
- super(tableNameWithType, streamConfigMap);
- }
-}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java
index aefa0e9093d..ae44bc086d8 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java
@@ -58,15 +58,10 @@ public class StreamConfigProperties {
* @deprecated because the property key is confusing (says size but is
actually rows). Use
* {@link StreamConfigProperties#SEGMENT_FLUSH_THRESHOLD_ROWS}
*
- * Row count flush threshold for realtime segments. This behaves in a
similar way for HLC and LLC. For HLC,
- * since there is only one consumer per server, this size is used as the
size of the consumption buffer and
- * determines after how many rows we flush to disk. For example, if this
threshold is set to two million rows,
- * then a high level consumer would have a buffer size of two million.
- *
- * For LLC, this size is divided across all the segments assigned to a given
server and is set on a per segment
- * basis. Assuming a low level consumer server is assigned four stream
partitions to consume from and a flush
- * size of two million, then each consuming segment would have a flush size
of five hundred thousand rows, for a
- * total of two million rows in memory.
+ * Row count flush threshold for realtime segments. This size is divided
across all the segments assigned to a given
+ * server and is set on a per segment basis. Assuming a server is assigned
four stream partitions to consume from and
+ * a flush size of two million, then each consuming segment would have a
flush size of five hundred thousand rows, for
+ * a total of two million rows in memory.
*
* Keep in mind that this NOT a hard threshold, as other tables can also be
assigned to this server, and that in
* certain conditions (eg. if the number of servers, replicas of partitions
changes) where partition
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
index 6502ea1a474..fb4de469073 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.spi.stream;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.pinot.spi.utils.retry.RetryPolicy;
@@ -62,30 +61,24 @@ public abstract class StreamConsumerFactory {
return new LongMsgOffsetFactory();
}
- /**
- * Creates a partition group consumer, which can fetch messages from a
partition group
- */
- public PartitionGroupConsumer createPartitionGroupConsumer(String clientId,
- PartitionGroupConsumptionStatus partitionGroupConsumptionStatus) {
- return createPartitionLevelConsumer(clientId,
partitionGroupConsumptionStatus.getStreamPartitionGroupId());
- }
+ /// Creates a [PartitionGroupConsumer] that fetches messages from the
partition group described by
+ /// `partitionGroupConsumptionStatus`.
+ ///
+ /// Every [StreamConsumerFactory] implementation must override this method.
The returned consumer is owned by
+ /// the caller, which is responsible for closing it via
[PartitionGroupConsumer#close()]. Each invocation returns
+ /// a new consumer instance; implementations are not required to make the
returned consumer thread-safe.
+ ///
+ /// @param clientId identifies the creator of this consumer
+ /// @param partitionGroupConsumptionStatus the partition group to consume
and the offsets to start from
+ /// @return a new, non-null partition group consumer
+ public abstract PartitionGroupConsumer createPartitionGroupConsumer(String
clientId,
+ PartitionGroupConsumptionStatus partitionGroupConsumptionStatus);
public PartitionGroupConsumer createPartitionGroupConsumer(String clientId,
PartitionGroupConsumptionStatus partitionGroupConsumptionStatus,
RetryPolicy retryPolicy) {
return createPartitionGroupConsumer(clientId,
partitionGroupConsumptionStatus);
}
- @Deprecated
- public PartitionLevelConsumer createPartitionLevelConsumer(String clientId,
int partition) {
- throw new UnsupportedOperationException();
- }
-
- @Deprecated
- public StreamLevelConsumer createStreamLevelConsumer(String clientId, String
tableName, Set<String> fieldsToRead,
- String groupId) {
- throw new UnsupportedOperationException();
- }
-
public static String getUniqueClientId(String prefix) {
if (prefix == null) {
return String.valueOf(CLIENT_ID_SEQ.getAndIncrement());
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamLevelConsumer.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamLevelConsumer.java
deleted file mode 100644
index 87f1493df58..00000000000
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamLevelConsumer.java
+++ /dev/null
@@ -1,62 +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.spi.stream;
-
-import org.apache.pinot.spi.annotations.InterfaceAudience;
-import org.apache.pinot.spi.annotations.InterfaceStability;
-import org.apache.pinot.spi.data.readers.GenericRow;
-
-
-/**
- * DEPRECATED - since = "Pinot no longer support high level consumer model
since v0.12.*"
- * Interface for a consumer that consumes at stream level and is unaware of
any partitions of the stream
- */
-@Deprecated
[email protected]
[email protected]
-public interface StreamLevelConsumer {
-
- /**
- * Initialize and start the stream level consumer
- * @throws Exception
- */
- void start()
- throws Exception;
-
- /**
- * Get next row from the stream and decode it into a generic row
- * @param destination
- * @return
- */
- GenericRow next(GenericRow destination);
-
- /**
- * Commit the offsets consumed so far
- * The next call to consume should exclude all events consumed before the
commit was called, and start from newer
- * events not yet consumed
- */
- void commit();
-
- /**
- * Shutdown the stream consumer
- * @throws Exception
- */
- void shutdown()
- throws Exception;
-}
diff --git
a/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java
b/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java
index e7a18b1d249..79d1cdadd76 100644
---
a/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java
+++
b/pinot-tools/src/main/java/org/apache/pinot/tools/utils/KafkaStarterUtils.java
@@ -53,8 +53,6 @@ public class KafkaStarterUtils {
public static final String KAFKA_PRODUCER_CLASS_NAME =
getKafkaConnectorPackageName() + ".server.KafkaDataProducer";
public static final String KAFKA_STREAM_CONSUMER_FACTORY_CLASS_NAME =
getKafkaConnectorPackageName() + ".KafkaConsumerFactory";
- public static final String KAFKA_STREAM_LEVEL_CONSUMER_CLASS_NAME =
- getKafkaConnectorPackageName() + ".KafkaStreamLevelConsumer";
public static final String KAFKA_JSON_MESSAGE_DECODER_CLASS_NAME =
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder";
diff --git
a/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json
index 55b201fa7cb..d3815947751 100644
---
a/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/minions/stream/githubEvents/githubEvents_realtime_table_config.json
@@ -28,7 +28,6 @@
"stream.kafka.topic.name": "githubEvents",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "1h",
"realtime.segment.flush.threshold.size": "100",
diff --git
a/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json
index 60877325156..7a1ab1def2d 100644
---
a/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/airlineStats/airlineStats_realtime_table_config.json
@@ -23,7 +23,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000"
diff --git
a/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json
index e936e8bb071..25794fbb598 100644
---
a/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/airlineStats/docker/airlineStats_realtime_table_config.json
@@ -18,8 +18,6 @@
"stream.kafka.topic.name": "flights-realtime",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.hlc.zk.connect.string": "zookeeper:2181/kafka",
- "stream.kafka.zk.broker.url": "zookeeper:2181/kafka",
"stream.kafka.broker.list": "kafka:9092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000",
diff --git
a/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json
index def6aefe84f..6ac0499a94f 100644
---
a/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/dailySales/dailySales_realtime_table_config.json
@@ -26,7 +26,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000"
diff --git
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
index 35acf8525c8..2fe1b6a0751 100644
---
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews/fineFoodReviews_realtime_table_config.json
@@ -31,7 +31,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000"
diff --git
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
index 56472db88a5..dd84b3e1752 100644
---
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_0/fineFoodReviews_part_0_realtime_table_config.json
@@ -32,7 +32,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000"
diff --git
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
index 80464a7caac..58e0967c856 100644
---
a/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/fineFoodReviews_part_1/fineFoodReviews_part_1_realtime_table_config.json
@@ -32,7 +32,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "3600000",
"realtime.segment.flush.threshold.size": "50000"
diff --git
a/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json
index 4c83abd311b..abcf6ed944f 100644
---
a/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/githubEvents/githubEvents_realtime_table_config.json
@@ -21,7 +21,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "1h",
"realtime.segment.flush.threshold.size": "100"
diff --git
a/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json
index a799acefff6..e0d2e080f48 100644
---
a/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/meetupRsvp/meetupRsvp_realtime_table_config.json
@@ -18,7 +18,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "largest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"stream.kafka.metadata.populate": "true"
}
diff --git
a/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json
index d5e00c02643..dc708946400 100644
---
a/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/meetupRsvpComplexType/meetupRsvpComplexType_realtime_table_config.json
@@ -17,7 +17,6 @@
"stream.kafka.topic.name": "meetupRSVPComplexTypeEvents",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
]
diff --git
a/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json
index a21dac74112..5faa2e1bc86 100644
---
a/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/meetupRsvpJson/meetupRsvpJson_realtime_table_config.json
@@ -30,7 +30,6 @@
"stream.kafka.topic.name": "meetupRSVPJsonEvents",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
]
diff --git
a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json
index 6ba48ea1e7c..0c242cc6a3c 100644
---
a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/docker/pullRequestMergedEvents_realtime_table_config.json
@@ -23,7 +23,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "pinot-zookeeper:2181/kafka",
"stream.kafka.broker.list": "kafka:9092",
"realtime.segment.flush.threshold.time": "12h",
"realtime.segment.flush.threshold.size": "100000"
diff --git
a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json
index 7e74a032db8..f87476afe43 100644
---
a/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/pullRequestMergedEvents/pullRequestMergedEvents_realtime_table_config.json
@@ -23,7 +23,6 @@
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092",
"realtime.segment.flush.threshold.time": "12h",
"realtime.segment.flush.threshold.size": "100000"
diff --git
a/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
index 73c0c595304..a5589b96703 100644
---
a/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/upsertJsonMeetupRsvp/upsertJsonMeetupRsvp_realtime_table_config.json
@@ -59,7 +59,6 @@
"stream.kafka.topic.name": "upsertJsonMeetupRSVPEvents",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
]
diff --git
a/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
index d5b9c62aaec..393b5fa7e47 100644
---
a/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/upsertMeetupRsvp/upsertMeetupRsvp_realtime_table_config.json
@@ -47,7 +47,6 @@
"stream.kafka.topic.name": "upsertMeetupRSVPEvents",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
],
diff --git
a/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
b/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
index 27a0cf70282..1cc953a95d0 100644
---
a/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
+++
b/pinot-tools/src/main/resources/examples/stream/upsertPartialMeetupRsvp/upsertPartialMeetupRsvp_realtime_table_config.json
@@ -47,7 +47,6 @@
"stream.kafka.topic.name": "upsertPartialMeetupRSVPEvents",
"stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
- "stream.kafka.zk.broker.url": "localhost:2191/kafka",
"stream.kafka.broker.list": "localhost:19092"
}
]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
