Merge branch 'master' into vertica_driver

Author: timbrownls20

docs/pages/product/caching/running-in-production.mdx

@@ -325,6 +325,28 @@ Refresh worker should be able to finish pre-aggregation refresh before
 garbage collection starts. It means that all pre-aggregation partitions
 should be built before any tables are removed.
 
+#### Supported file systems
+
+The garbage collection mechanism relies on the ability of the underlying file
+system to report the creation time of a file.
+
+If the file system does not support getting the creation time, you will see the
+following error message in Cube Store logs:
+
+```
+ERROR [cubestore::remotefs::cleanup] <pid:1>
+error while getting created time for file "<name>.chunk.parquet":
+creation time is not available for the filesystem
+```
+
+<ReferenceBox>
+
+XFS is known to not support getting the creation time of a file.
+Please see [this issue](https://github.com/cube-js/cube/issues/7905#issuecomment-2504212623)
+for possible workarounds.
+
+</ReferenceBox>
+
 ## Security
 
 ### Authentication

docs/pages/reference/data-model/pre-aggregations.mdx

@@ -1635,6 +1635,13 @@ cubes:
 
 </CodeTabs>
 
+<ReferenceBox>
+
+In some cases, indexes would not work with `original_sql` pre-aggregations.
+Please [track this issue](https://github.com/cube-js/cube/issues/7420).
+
+</ReferenceBox>
+
 #### `type`
 
 This option is used to define [aggregating indexes][ref-aggregating-indexes]

packages/cubejs-ksql-driver/src/KsqlDriver.ts

@@ -131,7 +131,9 @@ export class KsqlDriver extends BaseDriver implements DriverInterface {
     if (this.config.kafkaHost) {
       this.kafkaClient = new Kafka({
         clientId: 'Cube',
-        brokers: [this.config.kafkaHost],
+        brokers: this.config.kafkaHost
+          .split(',')
+          .map(h => h.trim()),
         // authenticationTimeout: 10000,
         // reauthenticationThreshold: 10000,
         ssl: this.config.kafkaUseSsl,

rust/cubestore/cubestore/src/streaming/kafka.rs

@@ -306,6 +306,13 @@ impl StreamingSource for KafkaStreamingSource {
         let unique_key_columns = self.unique_key_columns.clone();
         let seq_column_index_to_move = self.seq_column_index;
         let traffic_sender = TrafficSender::new(self.trace_obj.clone());
+        let hosts = self
+            .host
+            .clone()
+            .split(",")
+            .filter(|s| !s.is_empty())
+            .map(|s| s.trim().to_string())
+            .collect();
         let stream = self
             .kafka_client
             .create_message_stream(
@@ -321,7 +328,7 @@ impl StreamingSource for KafkaStreamingSource {
                         })
                         .unwrap_or(Offset::End),
                 ),
-                vec![self.host.clone()],
+                hosts,
                 &self.user,
                 &self.password,
                 self.use_ssl,