Skip to content
This repository has been archived by the owner on Oct 18, 2021. It is now read-only.

Increase meta client connection timeout。 #161

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ class MetaProvider(addresses: List[HostAndPort], timeout: Int, retry: Int)
address.append(new HostAddress(addr.getHostText, addr.getPort))
}

private val metaClient = new MetaClient(address.asJava)
private val metaClient = new MetaClient(address.asJava,timeout,retry,retry)
metaClient.connect()

def getPartNumber(space: String): Int = {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,10 @@ import com.vesoft.nebula.meta.{PropertyType, Schema}
import scala.collection.JavaConverters._
import scala.collection.mutable

class MetaProvider(addresses: List[Address]) extends AutoCloseable {
class MetaProvider(addresses: List[Address], timeout:Int, connectionRetry:Int, executionRetry:Int) extends AutoCloseable {

val metaAddress = addresses.map(address => new HostAddress(address._1, address._2)).asJava
val client = new MetaClient(metaAddress)
val client = new MetaClient(metaAddress,timeout,connectionRetry,executionRetry)
client.connect()

def getPartitionNumber(space: String): Int = {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ abstract class NebulaPartitionReader extends InputPartitionReader[InternalRow] {
this()
this.schema = schema

metaProvider = new MetaProvider(nebulaOptions.getMetaAddress)
metaProvider = new MetaProvider(nebulaOptions.getMetaAddress,nebulaOptions.timeout,nebulaOptions.connectionRetry,nebulaOptions.executionRetry)
val address: ListBuffer[HostAddress] = new ListBuffer[HostAddress]

for (addr <- nebulaOptions.getMetaAddress) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ abstract class NebulaSourceReader(nebulaOptions: NebulaOptions) extends DataSour
val returnCols = nebulaOptions.getReturnCols
val noColumn = nebulaOptions.noColumn
val fields: ListBuffer[StructField] = new ListBuffer[StructField]
val metaProvider = new MetaProvider(nebulaOptions.getMetaAddress)
val metaProvider = new MetaProvider(nebulaOptions.getMetaAddress,nebulaOptions.timeout,nebulaOptions.connectionRetry,nebulaOptions.executionRetry)

import scala.collection.JavaConverters._
var schemaCols: Seq[ColumnDef] = Seq()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ class NebulaWriter(nebulaOptions: NebulaOptions) extends Serializable {
val failedExecs: ListBuffer[String] = new ListBuffer[String]

val graphProvider = new GraphProvider(nebulaOptions.getGraphAddress)
val metaProvider = new MetaProvider(nebulaOptions.getMetaAddress)
val metaProvider = new MetaProvider(nebulaOptions.getMetaAddress,nebulaOptions.timeout,nebulaOptions.connectionRetry,nebulaOptions.executionRetry)
val isVidStringType = metaProvider.getVidType(nebulaOptions.spaceName) == VidType.STRING

def prepareSpace(): Unit = {
Expand Down