Offset operations-DataHub(DataHub)-阿里云帮助中心

Initialize offset

Parameters

Name	Type	Description
projectName	string	The project name.
topicName	string	The topic name.
subId	string	The ID of the subscription.
shardIds	string	The list of shard IDs.

Response example

type OpenSubscriptionSessionResult struct {
    Offsets map[string]SubscriptionOffset `json:"Offsets"`
}
// SubscriptionOffset
type SubscriptionOffset struct {
    Timestamp int64  `json:"Timestamp"`
    Sequence  int64  `json:"Sequence"`
    VersionId int64  `json:"Version"`
    SessionId *int64 `json:"SessionId"`
}

Errors

Class name	Error code	Description
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The requested resource does not exist. This error may occur if you send a request immediately after a Split or Merge operation.
AuthorizationFailedError	`Unauthorized`	The authorization signature could not be parsed. Verify that the AccessKey is valid.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	Invalid parameters.
DatahubClientError	-	The base class for all other exceptions.

Code example

func openOffset(dh datahub.DataHub, projectName, topicName string) {
    subId := "1565580329258VXSY8"
    shardIds := []string{"0", "1", "2"}
    oss, err := dh.OpenSubscriptionSession(projectName, topicName, subId, shardIds)
    if err != nil {
        fmt.Println("open session failed")
        fmt.Println(err)
        return
    }
    fmt.Println("open session successful")
    fmt.Println(oss)
}

Get offset

Retrieves the current offset information of a subscription. Unlike OpenSubscriptionSession, the SessionId in the SubscriptionOffset returned by GetSubscriptionOffset is nil, so you cannot commit the offset. Use this method only to view offset information.

Parameters

Name	Type	Description
projectName	string	The project name.
topicName	string	The topic name.
subId	string	The ID of the subscription.
shardIds	string	The list of shard IDs.

Response example

type OpenSubscriptionSessionResult struct {
    Offsets map[string]SubscriptionOffset `json:"Offsets"`
}
// SubscriptionOffset
type SubscriptionOffset struct {
    Timestamp int64  `json:"Timestamp"`
    Sequence  int64  `json:"Sequence"`
    VersionId int64  `json:"Version"`
    SessionId *int64 `json:"SessionId"`
}

Errors

Class name	Error code	Description
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The requested resource does not exist. This error may occur if you send a request immediately after a Split or Merge operation.
AuthorizationFailedError	`Unauthorized`	The authorization signature could not be parsed. Verify that the AccessKey is valid.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	Invalid parameters.
DatahubClientError	-	The base class for all other exceptions.

Code example

func getOffset(dh datahub.DataHub, projectName, topicName string) {
    subId := "1565580329258VXSY8"
    shardIds := []string{"0", "1", "2"}
    gss, err := dh.GetSubscriptionOffset(projectName, topicName, subId, shardIds)
    if err != nil {
        fmt.Println("get session failed")
        fmt.Println(err)
        return
    }
    fmt.Println("get session successful")
    fmt.Println(gss)
}

Update offset

Parameters

Parameter	Type	Description
projectName	string	The project name.
topicName	string	The topic name.
subId	string	The ID of the subscription.
offsets	map[string]SubscriptionOffset	The offset map of the shards.

Response example

Errors

Class name	Error code	Description
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The requested resource does not exist. This error may occur if you send a request immediately after a Split or Merge operation.
AuthorizationFailedError	`Unauthorized`	The authorization signature could not be parsed. Verify that the AccessKey is valid.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	Invalid parameters.
DatahubClientError	-	The base class for all other exceptions.

Code example

func updateOffset() {
    shardIds := []string{"0", "1", "2"}
    oss, err := dh.OpenSubscriptionSession(projectName, topicName, subId, shardIds)
    if err != nil {
        fmt.Println("open session failed")
        fmt.Println(err)
    }
    fmt.Println("open session successful")
    fmt.Println(oss)
    offset := oss.Offsets["0"]
    // set offset message
    offset.Sequence = 900
    offset.Timestamp = 1565593166690
    offsetMap := map[string]datahub.SubscriptionOffset{
        "0": offset,
    }
    if err := dh.CommitSubscriptionOffset(projectName, topicName, subId, offsetMap); err != nil {
        if _, ok := err.(*datahub.SubscriptionOfflineError); ok {
            fmt.Println("the subscription has offline")
        } else if _, ok := err.(*datahub.SubscriptionSessionInvalidError); ok {
            fmt.Println("the subscription is open elsewhere")
        } else if _, ok := err.(*datahub.SubscriptionOffsetResetError); ok {
            fmt.Println("the subscription is reset elsewhere")
        } else {
            fmt.Println(err)
        }
        fmt.Println("update offset failed")
        return
    }
    fmt.Println("update offset successful")
}

Reset offset

Resetting an offset sets it to a specific timestamp. After the reset, the VersionId in the offset information increases by 1, and the previous session becomes invalid and cannot be used to update the offset.

Parameters

Parameter	Type	Description
projectName	string	The project name.
topicName	string	The topic name.
subId	string	The ID of the subscription.
offsets	map[string]SubscriptionOffset	The offset map of the shards.

Response example

Errors

Class name	Error code	Description
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The requested resource does not exist. This error may occur if you send a request immediately after a Split or Merge operation.
AuthorizationFailedError	`Unauthorized`	The authorization signature could not be parsed. Verify that the AccessKey is valid.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	Invalid parameters.
DatahubClientError	-	The base class for all other exceptions.

Code example

func resetOffset(dh datahub.DataHub, projectName, topicName string) {
    subId := "1565580329258VXSY8"
    offset := datahub.SubscriptionOffset{
        Timestamp: 1565593166690,
    }
    offsetMap := map[string]datahub.SubscriptionOffset{
        "1": offset,
    }
    if err := dh.ResetSubscriptionOffset(projectName, topicName, subId, offsetMap); err != nil {
        fmt.Println("reset offset failed")
        fmt.Println(err)
        return
    }
    fmt.Println("reset offset successful")
}