Queues MV: Show query compile error (#456)

* Queues MV: Show query compile error * Kafka: Explicit nullable declaration on the rebalance callback --------- Co-authored-by: djklim87 <[email protected]>
manticoresoftware · Jan 17, 2025 · 3a3d522 · 3a3d522
1 parent ed387a3
commit 3a3d522
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 3 deletions.
diff --git a/src/Plugin/Queue/Handlers/View/CreateViewHandler.php b/src/Plugin/Queue/Handlers/View/CreateViewHandler.php
@@ -18,6 +18,7 @@
 use Manticoresearch\Buddy\Core\Plugin\BaseHandlerWithClient;
 use Manticoresearch\Buddy\Core\Task\Task;
 use Manticoresearch\Buddy\Core\Task\TaskResult;
+use Manticoresearch\Buddy\Core\Tool\Buddy;
 use PHPSQLParser\PHPSQLCreator;
 use PHPSQLParser\exceptions\UnsupportedFeatureException;
 
@@ -180,9 +181,10 @@ public function run(): Task {
 	 * @param int $iterations
 	 * @param int $startFrom
 	 * @param int $suspended
+	 *
 	 * @return array<string, array<string, string>>
 	 * @throws ManticoreSearchClientError
-	 * @throws UnsupportedFeatureException
+	 * @throws UnsupportedFeatureException|GenericError
 	 */
 	public static function createViewRecords(
 		Client $client,
@@ -206,7 +208,16 @@ public static function createViewRecords(
 			$parsedQuery['FROM'][0]['no_quotes']['parts'] = [$bufferTableName];
 			$parsedQuery['FROM'][0]['base_expr'] = $bufferTableName;
 
-			$query = (new PHPSQLCreator())->create($parsedQuery);
+			$query = '';
+			try {
+				$query = (new PHPSQLCreator())->create($parsedQuery);
+			} catch (\Exception $exception) {
+				$message = "Can\'t compile SELECT query from ".json_encode($parsedQuery);
+				Buddy::debugv($message);
+				Buddy::debugv($exception->getMessage());
+				GenericError::throw($message);
+			}
+
 			$escapedQuery = str_replace("'", "\\'", $query);
 			$escapedOriginalQuery = str_replace("'", "\\'", $originalQuery);
 

diff --git a/src/Plugin/Queue/Workers/Kafka/KafkaWorker.php b/src/Plugin/Queue/Workers/Kafka/KafkaWorker.php
@@ -105,7 +105,7 @@ private function initKafkaConfig(): Conf {
 
 		// Set a rebalance callback to log partition assignments (optional)
 		$conf->setRebalanceCb(
-			function (KafkaConsumer $kafka, $err, array $partitions = null) {
+			function (KafkaConsumer $kafka, $err, ?array $partitions = null) {
 				switch ($err) {
 					case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
 						$kafka->assign($partitions);