diff --git a/src/main/java/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnectorConfig.java b/src/main/java/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnectorConfig.java index 359e35b..a086808 100644 --- a/src/main/java/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnectorConfig.java +++ b/src/main/java/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnectorConfig.java @@ -28,8 +28,6 @@ class CloudSolrSinkConnectorConfig extends SolrSinkConnectorConfig { public static final String ZOOKEEPER_CONNECT_TIMEOUT_CONFIG = "solr.zookeeper.connect.timeout.ms"; public static final String ZOOKEEPER_CLIENT_TIMEOUT_CONFIG = "solr.zookeeper.client.timeout.ms"; public static final String ZOOKEEPER_RETRY_EXPIRY_TIME_CONFIG = "solr.zookeeper.retry.expiry.time.ms"; - public static final String SOLR_CONNECT_TIMEOUT_CONFIG = "solr.connect.timeout.ms"; - public static final String SOLR_SOCKET_TIMEOUT_CONFIG = "solr.socket.timeout.ms"; private static final String ZOOKEEPER_HOSTS_DOC = "Zookeeper hosts that are used to store solr configuration."; @@ -38,16 +36,12 @@ class CloudSolrSinkConnectorConfig extends SolrSinkConnectorConfig { private static final String ZOOKEEPER_CLIENT_TIMEOUT_DOC = "Set the timeout to the zookeeper ensemble in ms."; private static final String ZOOKEEPER_RETRY_EXPIRY_TIME_DOC = "This is the time to wait to refetch the " + "state after getting the same state version from ZK in ms."; - private static final String SOLR_CONNECT_TIMEOUT_DOC = "Set the connect timeout to the solr in ms."; - private static final String SOLR_SOCKET_TIMEOUT_DOC = "Set the solr read timeout on all sockets in ms."; public final List<String> zookeeperHosts; public final String zookeeperChroot; public final int zookeeperConnectTimeoutMs; public final int zookeeperClientTimeoutMs; public final int zookeeperRetryExpiryTimeMs; - public final int solrConnectTimeoutMs; - public final int solrSocketTimeoutMs; protected CloudSolrSinkConnectorConfig(Map<String, String> props) { super(config(), props); @@ -56,8 +50,6 @@ protected CloudSolrSinkConnectorConfig(Map<String, String> props) { this.zookeeperConnectTimeoutMs = getInt(ZOOKEEPER_CONNECT_TIMEOUT_CONFIG); this.zookeeperClientTimeoutMs = getInt(ZOOKEEPER_CLIENT_TIMEOUT_CONFIG); this.zookeeperRetryExpiryTimeMs = getInt(ZOOKEEPER_RETRY_EXPIRY_TIME_CONFIG); - this.solrConnectTimeoutMs = getInt(SOLR_CONNECT_TIMEOUT_CONFIG); - this.solrSocketTimeoutMs = getInt(SOLR_SOCKET_TIMEOUT_CONFIG); } @@ -97,20 +89,6 @@ public static ConfigDef config() { .group(CONNECTION_GROUP) .defaultValue(3000) .build() - ).define( - ConfigKeyBuilder.of(SOLR_CONNECT_TIMEOUT_CONFIG, ConfigDef.Type.INT) - .importance(ConfigDef.Importance.LOW) - .documentation(SOLR_CONNECT_TIMEOUT_DOC) - .group(CONNECTION_GROUP) - .defaultValue(15000) - .build() - ).define( - ConfigKeyBuilder.of(SOLR_SOCKET_TIMEOUT_CONFIG, ConfigDef.Type.INT) - .importance(ConfigDef.Importance.LOW) - .documentation(SOLR_SOCKET_TIMEOUT_DOC) - .group(CONNECTION_GROUP) - .defaultValue(120000) - .build() ); } } diff --git a/src/main/java/com/github/jcustenborder/kafka/connect/solr/SolrSinkConnectorConfig.java b/src/main/java/com/github/jcustenborder/kafka/connect/solr/SolrSinkConnectorConfig.java index afc87a3..834dc43 100644 --- a/src/main/java/com/github/jcustenborder/kafka/connect/solr/SolrSinkConnectorConfig.java +++ b/src/main/java/com/github/jcustenborder/kafka/connect/solr/SolrSinkConnectorConfig.java @@ -27,6 +27,8 @@ class SolrSinkConnectorConfig extends AbstractConfig { public static final String SOLR_USERNAME_CONFIG = "solr.username"; public static final String SOLR_PASSWORD_CONFIG = "solr.password"; public static final String SOLR_DELETE_DOCUMENTS_CONFIG = "solr.delete.documents.enabled"; + public static final String SOLR_CONNECT_TIMEOUT_CONFIG = "solr.connect.timeout.ms"; + public static final String SOLR_SOCKET_TIMEOUT_CONFIG = "solr.socket.timeout.ms"; static final String SOLR_USERNAME_DOC = "The username to use for basic authentication."; static final String SOLR_PASSWORD_DOC = "The password to use for basic authentication."; @@ -35,12 +37,16 @@ class SolrSinkConnectorConfig extends AbstractConfig { "the standard Solr commit setting."; static final String SOLR_DELETE_DOCUMENTS_DOC = "Flag to determine if the connector should delete documents. General " + "practice in Kafka is to treat a record that contains a key with a null value as a delete."; + static final String SOLR_CONNECT_TIMEOUT_DOC = "Set the connect timeout to the solr in ms."; + static final String SOLR_SOCKET_TIMEOUT_DOC = "Set the solr read timeout on all sockets in ms."; public final String username; public final String password; public final boolean useBasicAuthentication; public final int commitWithin; public final boolean deleteDocuments; + public final int solrConnectTimeoutMs; + public final int solrSocketTimeoutMs; protected SolrSinkConnectorConfig(ConfigDef configDef, Map<String, String> props) { @@ -50,6 +56,8 @@ protected SolrSinkConnectorConfig(ConfigDef configDef, Map<String, String> props this.password = this.getPassword(SOLR_PASSWORD_CONFIG).value(); this.useBasicAuthentication = !Strings.isNullOrEmpty(this.username); this.deleteDocuments = this.getBoolean(SOLR_DELETE_DOCUMENTS_CONFIG); + this.solrConnectTimeoutMs = this.getInt(SOLR_CONNECT_TIMEOUT_CONFIG); + this.solrSocketTimeoutMs = this.getInt(SOLR_SOCKET_TIMEOUT_CONFIG); } public static final String AUTHENTICATION_GROUP = "Authentication"; @@ -89,6 +97,20 @@ public static ConfigDef config() { .documentation(SOLR_DELETE_DOCUMENTS_DOC) .group(INDEXING_GROUP) .build() - ); + ).define( + ConfigKeyBuilder.of(SOLR_CONNECT_TIMEOUT_CONFIG, ConfigDef.Type.INT) + .importance(ConfigDef.Importance.LOW) + .documentation(SOLR_CONNECT_TIMEOUT_DOC) + .group(CONNECTION_GROUP) + .defaultValue(15000) + .build() + ).define( + ConfigKeyBuilder.of(SOLR_SOCKET_TIMEOUT_CONFIG, ConfigDef.Type.INT) + .importance(ConfigDef.Importance.LOW) + .documentation(SOLR_SOCKET_TIMEOUT_DOC) + .group(CONNECTION_GROUP) + .defaultValue(120000) + .build() + ); } }