Java SDK usage guide

Initialization

To access DataHub, provide your Alibaba Cloud AccessKey ID and AccessKey, along with the DataHub endpoint. The following code creates a DataHubClient using the DataHub endpoint list:

SDK 2.25.1 and later (recommended)

//Create a DataHubClient instance using the new Batch transmission protocol
DatahubConfig.Protocol protocol = DatahubConfig.Protocol.BATCH;
DatahubClient datahubClient = DatahubClientBuilder.newBuilder().setDatahubConfig(
  		         //Protocol is optional. If you do not set this parameter, the PROTOBUF transmission protocol is used by default.
                new DatahubConfig(endpoint, new AliyunAccount(accessId, accessKey), protocol)
        ).setHttpConfig(new HttpConfig().setCompressType(CompressType.ZSTD)).build();

• Configuration description

  • DatahubConfig

    Name	Description
    endpoint	The endpoint of DataHub.
    account	Your Alibaba Cloud account information.
    protocol	The transmission protocol. Valid values: PROTOBUF and BATCH.

    Note

    To use the BATCH transmission protocol, you must enable multiple versions. You can enable multiple versions in the console or using the SDK. For more information, see DataHub cost-saving strategies.

  • HttpConfig

    Name	Description
    readTimeout	The socket read/write timeout period. Default value: 10s.
    connTimeout	The TCP connection timeout period. Default value: 10s.
    maxRetryCount	The maximum number of retries after a request failure. Default value: 1. Do not modify this parameter. Retries should be handled by the upper business layer.
    debugRequest	Specifies whether to print request logs. Default value: false.
    compressType	The compression method for data transmission. Default value: lz4. Valid values: lz4, deflate, and ztsd.
    proxyUri	The proxy server IP address.
    proxyUsername	The proxy server username.
    proxyPassword	The proxy server password.

• SDK statistics information

  The SDK collects request statistics such as queries per second (QPS) for put and get operations. Enable this feature with the following code:

  ClientMetrics.startMetrics();

  By default, metrics are printed to log files. Configure slf4j to enable logging. The metric package is: com.aliyun.datahub.client.metrics.

Write data to DataHub

The following example writes data to a tuple topic:

  public void writeTupleTopic(int maxRetry) {
    String shardId = "9";
    // Generate 10 records
    List<RecordEntry> recordEntries = new ArrayList<>();
    for (int i = 0; i < 10; ++i) {
      RecordEntry recordEntry = new RecordEntry();
      // Set additional attributes for each record
      recordEntry.addAttribute("key1", "value11");
      TupleRecordData data = new TupleRecordData(this.recordSchema);
      data.setField("field1", "Hello World");
      data.setField("field2", 1234567);
      recordEntry.setRecordData(data);
      recordEntry.setShardId(shardId);
      recordEntries.add(recordEntry);
    }

    int retryNum = 0;
    while (retryNum < maxRetry) {
      try {
        // Supported by the server from version 2.12. For earlier versions, use the putRecords interface
        //datahubClient.putRecordsByShard(Constant.projectName, Constant.topicName, shardId, recordEntries);
        PutRecordsResult putRecordsResult = this.datahubClient.putRecords(Constant.projectName,
            Constant.topicName, recordEntries);
        System.out.println("write tuple data successful");
        System.out.println(putRecordsResult.getPutErrorEntries());
        break;
      } catch (InvalidParameterException e) {
        // invalid parameter
        e.printStackTrace();
        throw e;
      } catch (AuthorizationFailureException e) {
        // AK error
        e.printStackTrace();
        throw e;
      } catch (ResourceNotFoundException e) {
        // project or topic not found
        e.printStackTrace();
        throw e;
      } catch (ShardSealedException e) {
        // shard status is CLOSED, read only
        e.printStackTrace();
        throw e;
      } catch (LimitExceededException e) {
        // limit exceed, retry
        e.printStackTrace();
        retryNum++;
      } catch (DatahubClientException e) {
        // other error
        e.printStackTrace();
        retryNum++;
      }
    }
  }

Create a subscription to consume DataHub data

//Example of offset consumption and offset commit during consumption
public static void example() {
  String shardId = "0";
  List<String> shardIds = Arrays.asList("0", "1");
  OpenSubscriptionSessionResult openSubscriptionSessionResult = datahubClient.openSubscriptionSession(Constant.projectName, Constant.topicName, subId, shardIds);
  SubscriptionOffset subscriptionOffset = openSubscriptionSessionResult.getOffsets().get(shardId);
  // 1. Obtain the cursor of the current offset. If the current offset has expired, obtain the cursor of the first record within the lifecycle. If the data is not consumed, also obtain the cursor of the first record within the lifecycle.
  String cursor = null;
  //sequence < 0 indicates that the data is not consumed
  if (subscriptionOffset.getSequence() < 0) {
      // Obtain the cursor of the first record within the lifecycle
      cursor = datahubClient.getCursor(Constant.projectName, Constant.topicName, shardId, CursorType.OLDEST).getCursor();
  } else {
      // Obtain the cursor of the next record
      long nextSequence = subscriptionOffset.getSequence() + 1;
      try {
          //When you obtain a cursor by SEQUENCE, a SeekOutOfRange error may be returned, indicating that the data of the current cursor has expired
          cursor = datahubClient.getCursor(Constant.projectName, Constant.topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();
      } catch (SeekOutOfRangeException e) {
          // Obtain the cursor of the first record within the lifecycle
          cursor = datahubClient.getCursor(Constant.projectName, Constant.topicName, shardId, CursorType.OLDEST).getCursor();
      }
  }
  // 2. Read and save the offset. The following example shows how to read tuple data and save the offset every 1,000 records
  long recordCount = 0L;
  // Read 10 records each time
  int fetchNum = 10;
  while (true) {
      try {
          GetRecordsResult getRecordsResult = datahubClient.getRecords(Constant.projectName, Constant.topicName, shardId, schema, cursor, fetchNum);
          if (getRecordsResult.getRecordCount() <= 0) {
              // No data. Sleep and then read
              Thread.sleep(1000);
              continue;
          }
          for (RecordEntry recordEntry : getRecordsResult.getRecords()) {
              //Consume data
              TupleRecordData data = (TupleRecordData) recordEntry.getRecordData();
              System.out.println("field1:" + data.getField("field1") + "\t"
                      + "field2:" + data.getField("field2"));
              // After the data is processed, set the offset
              ++recordCount;
              subscriptionOffset.setSequence(recordEntry.getSequence());
              subscriptionOffset.setTimestamp(recordEntry.getSystemTime());
              if (recordCount % 1000 == 0) {
        //Commit the offset
                  Map<String, SubscriptionOffset> offsetMap = new HashMap<>();
                  offsetMap.put(shardId, subscriptionOffset);
                  datahubClient.commitSubscriptionOffset(Constant.projectName, Constant.topicName, subId, offsetMap);
                  System.out.println("commit offset successful");
              }
          }
          cursor = getRecordsResult.getNextCursor();
      } catch (SubscriptionOfflineException | SubscriptionSessionInvalidException e) {
          // Exit. Offline: The subscription is offline. SubscriptionSessionInvalid: The subscription is being consumed by another client at the same time
          break;
      } catch (SubscriptionOffsetResetException e) {
          // The offset is reset. Obtain the SubscriptionOffset information again. The following example shows how to reset by sequence
          // If the offset is reset by timestamp, you need to obtain the cursor using CursorType.SYSTEM_TIME
          subscriptionOffset = datahubClient.getSubscriptionOffset(Constant.projectName, Constant.topicName, subId, shardIds).getOffsets().get(shardId);
          long nextSequence = subscriptionOffset.getSequence() + 1;
          cursor = datahubClient.getCursor(Constant.projectName, Constant.topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();
      } catch (DatahubClientException e) {
          // TODO: Determine whether to exit based on different exceptions
      } catch (Exception e) {
          break;
      }
  }
}

Exception types

The Java SDK (version 2.12 and later) provides well-organized exception types. Use the try-catch mechanism to capture and handle these exceptions.

Only DatahubClientException and LimitExceededException are retryable. DatahubClientException covers transient errors such as server busy and server unavailable. Add retry logic for these two exception types, but strictly limit the number of retries.

The following table lists the exceptions available in version 2.12 and later. The package path is com.aliyun.datahub.client.exception.