fix(hive): support JDBC URLs and HTTP noSasl connections
This commit is contained in:
12 files changed
+517
-67
No files matched your search
@@ -629,6 +629,31 @@ fn append_agent_url_params(base: String, params: Option<&str>) -> String {
|
||||
format!("{base}{separator}{params}")
|
||||
}
|
||||
|
||||
pub fn hive_uses_zookeeper_discovery(config: &ConnectionConfig) -> bool {
|
||||
if !matches!(config.db_type, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if config.url_params.as_deref().is_some_and(hive_parameters_use_zookeeper_discovery) {
|
||||
return true;
|
||||
}
|
||||
|
||||
config.host.trim().is_empty()
|
||||
&& config.connection_string.as_deref().is_some_and(hive_parameters_use_zookeeper_discovery)
|
||||
}
|
||||
|
||||
fn hive_parameters_use_zookeeper_discovery(source: &str) -> bool {
|
||||
source.split([';', '?', '#', '&']).any(|part| {
|
||||
let Some((key, value)) = part.split_once('=') else {
|
||||
return false;
|
||||
};
|
||||
let key = percent_decode_str(key.trim()).decode_utf8_lossy();
|
||||
let value = percent_decode_str(value.trim()).decode_utf8_lossy();
|
||||
key.eq_ignore_ascii_case("serviceDiscoveryMode")
|
||||
&& (value.eq_ignore_ascii_case("zookeeper") || value.eq_ignore_ascii_case("zookeeperha"))
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -716,6 +741,25 @@ mod tests {
|
||||
assert_eq!(params["sessionRole"], "metadata");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn structured_hive_uses_current_url_params_for_zookeeper_detection() {
|
||||
let mut cfg = config(DatabaseType::Hive, Some("default"));
|
||||
cfg.connection_string = Some(
|
||||
"jdbc:hive2://zk1.example.com:2181,zk2.example.com:2181/default;serviceDiscoveryMode=zooKeeper".to_string(),
|
||||
);
|
||||
assert!(!hive_uses_zookeeper_discovery(&cfg));
|
||||
|
||||
cfg.url_params = Some("serviceDiscoveryMode=zooKeeperHA;zooKeeperNamespace=hs2".to_string());
|
||||
assert!(hive_uses_zookeeper_discovery(&cfg));
|
||||
|
||||
cfg.url_params = Some("note=serviceDiscoveryMode=zooKeeper".to_string());
|
||||
assert!(!hive_uses_zookeeper_discovery(&cfg));
|
||||
|
||||
cfg.host.clear();
|
||||
cfg.url_params = None;
|
||||
assert!(hive_uses_zookeeper_discovery(&cfg));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mongodb_database_falls_back_to_uri_database() {
|
||||
let mut cfg = config(DatabaseType::MongoDb, None);
|
||||
|
||||
@@ -9,10 +9,10 @@ use mysql_async::prelude::Queryable;
|
||||
use mysql_async::Row as MysqlRow;
|
||||
|
||||
use crate::agent_connection::{
|
||||
agent_connect_params, agent_connect_params_with_role, h2_file_path_from_jdbc_url, is_h2_file_connection,
|
||||
mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver, oracle_alternate_connect_config_labels,
|
||||
oracle_alternate_connect_configs, oracle_error_with_driver_hint, should_retry_mongo_with_legacy_driver,
|
||||
trino_like_jdbc_connection_string, AgentSessionRole,
|
||||
agent_connect_params, agent_connect_params_with_role, h2_file_path_from_jdbc_url, hive_uses_zookeeper_discovery,
|
||||
is_h2_file_connection, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver,
|
||||
oracle_alternate_connect_config_labels, oracle_alternate_connect_configs, oracle_error_with_driver_hint,
|
||||
should_retry_mongo_with_legacy_driver, trino_like_jdbc_connection_string, AgentSessionRole,
|
||||
};
|
||||
use crate::agent_manager::{AgentManager, JavaRuntimeMode, DEFAULT_JRE_KEY};
|
||||
use crate::agent_recovery::{RecoveryDecision, RecoveryPolicy, RecoveryScope};
|
||||
@@ -2556,6 +2556,9 @@ impl AppState {
|
||||
// through one local tunnel endpoint would silently break Oracle Net routing.
|
||||
return Err("Oracle TNS connections cannot be combined with SSH, proxy, or HTTP tunnel layers. Remove the transport layer or use Service Name/SID mode.".to_string());
|
||||
}
|
||||
if hive_uses_zookeeper_discovery(config) {
|
||||
return Err("Hive ZooKeeper service discovery cannot be combined with SSH, proxy, or HTTP tunnel layers because discovered HiveServer2 nodes would bypass the configured transport. Remove the transport layer or use a direct HiveServer2 host and port.".to_string());
|
||||
}
|
||||
|
||||
#[cfg(feature = "mq-admin")]
|
||||
if config.db_type == DatabaseType::MessageQueue
|
||||
@@ -8078,6 +8081,23 @@ for line in sys.stdin:
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hive_zookeeper_discovery_rejects_static_transport_layers() {
|
||||
let (state, dir) = test_app_state().await;
|
||||
let mut config = mysql_config(Some("default"));
|
||||
config.db_type = DatabaseType::Hive;
|
||||
config.connection_string = Some(
|
||||
"jdbc:hive2://zk1.example.com:2181,zk2.example.com:2181/default;serviceDiscoveryMode=zooKeeper".to_string(),
|
||||
);
|
||||
config.url_params = Some("serviceDiscoveryMode=zooKeeper".to_string());
|
||||
config.transport_layers = vec![TransportLayerConfig::Ssh(ssh_layer("hive-zk-tunnel", ""))];
|
||||
|
||||
let error = state.connection_host_port("hive-zookeeper", &config).await.unwrap_err();
|
||||
assert!(error.contains("discovered HiveServer2 nodes would bypass the configured transport"));
|
||||
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolved_transport_layers_substitutes_shared_profiles() {
|
||||
let (state, dir) = test_app_state().await;
|
||||
|
||||
Reference in new issue
Block a user