{
  "apiKey": 38,
  "type": "response",
  "name": "CreateDelegationTokenResponse",
  // Starting in version 1, on quota violation, brokers send out responses before throttling.
  "validVersions": "0-2on quota violation, brokers send out responses before throttling.
  //
  // Version 2 is the first flexible version.
  //
  // Version 3 adds the token requester principal
  "validVersions": "0-3",
  "flexibleVersions": "2+",
  "fields": [
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The top-level error, or zero if there was no error."},
    { "name": "OwnerPrincipalType", "type": "CreatableOwnerstring", "versions": "20+",
      "about": "OwnerThe is an Kafka PrincipalType+name string, who is the owner principal type of the token owner." },
      "fields": [
        { "name": "PrincipalTypePrincipalName", "type": "string", "versions": "20+",
          "about": "The typename of the Kafkatoken principalowner." },
        { "name": "PrincipalNameTokenRequesterPrincipalType", "type": "string", "versions": "23+",
          "about": "The nameprincipal type of the Kafka principalrequester of the token." }
      ]},
    { "name": "TokenRequesterTokenRequesterPrincipalName", "type": "CreatableTokenRequesterstring", "versions": "23+",
      "about": "TokenThe requesterprincipal istype anof Kafkathe PrincipalType+name string, who requested thisrequester of the token.", "fields": [
  },
     { "name": "PrincipalTypeIssueTimestampMs", "type": "stringint64", "versions": "20+",
        "about": "TheWhen typethis oftoken thewas Kafka principalgenerated." },
      { "name": "PrincipalNameExpiryTimestampMs", "type": "stringint64", "versions": "20+",
        "about": "TheWhen namethis of the Kafka principaltoken expires." }
    ]},
    { "name": "IssueTimestampMsMaxTimestampMs", "type": "int64", "versions": "0+",
      "about": "WhenThe maximum lifetime of this token was generated." },
    { "name": "ExpiryTimestampMsTokenId", "type": "int64string", "versions": "0+",
      "about": "WhenThe this token expiresUUID." },
    { "name": "MaxTimestampMsHmac", "type": "int64bytes", "versions": "0+",
      "about": "TheHMAC maximumof lifetimethe of thisdelegation token." },
    { "name": "TokenIdThrottleTimeMs", "type": "stringint32", "versions": "0+",
      "about": "The token UUID." },
    { "name": "Hmac", "type": "bytes", "versions": "0+",
      "about": "HMAC of the delegation token." },
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }
  ]
}

{
  "apiKey": 41,
  "type": "response",
  "name": "DescribeDelegationTokenResponse",
  // Starting in version 1, on quota violation, brokers send out responses before throttling.
  "validVersions": "0-2",
  "fields": [
    { "name": "ErrorCodeTokens", "type": "int16[]DescribedDelegationToken", "versions": "0+",
      "about": "The error codetokens.", or 0 if there was no error." },
    "fields": [
      { "name": "TokensPrincipalType", "type": "[]DescribedDelegationTokenstring", "versions": "0+",
        "about": "The tokenstoken principal type.", "fields": [ },
      { "name": "OwnerPrincipalName", "type": "CreatableOwnerstring", "versions": "20+",
        "about": "OwnerThe istoken an Kafka PrincipalType+name string, who is the owner of the token."principal name." },
        "fields": [
          { "name": "PrincipalTypeTokenRequesterPrincipalType", "type": "string", "versions": "23+",
            "about": "The principal type of the requester Kafkaof the principaltoken." },
          { "name": "PrincipalNameTokenRequesterPrincipalName", "type": "string", "versions": "23+",
            ""about": "The nameprincipal type of the requester Kafkaof the principaltoken." }
        ]},
      { "name": "TokenRequesterIssueTimestamp", "type": "CreatableTokenRequesterint64", "versions": "20+",
        "about": "TokenThe requestertoken isissue antimestamp Kafka PrincipalType+name string, who requested this token", "fields": [
        { "name": "PrincipalType"in milliseconds." },
      { "name": "ExpiryTimestamp", "type": "stringint64", "versions": "20+",
          "about": "The typetoken ofexpiry thetimestamp Kafkain principalmilliseconds." },
        { "name": "PrincipalNameMaxTimestamp", "type": "stringint64", "versions": "20+",
          "about": "The nametoken maximum oftimestamp thelength Kafkain principalmilliseconds." }
      ]},
      { "name": "IssueTimestampTokenId", "type": "int64string", "versions": "0+",
        "about": "The token issue timestamp in millisecondsID." },
      { "name": "ExpiryTimestampHmac", "type": "int64bytes", "versions": "0+",
        "about": "The token expiry timestamp in millisecondsHMAC." },
      { "name": "MaxTimestampRenewers", "type": "int64[]DescribedDelegationTokenRenewer", "versions": "0+",
        "about": "The token maximum timestamp length in milliseconds." },Those who are able to renew this token before it expires.", "fields": [
        { "name": "TokenIdPrincipalType", "type": "string", "versions": "0+",
          "about": "The tokenrenewer ID." principal type" },
        { "name": "HmacPrincipalName", "type": "bytesstring", "versions": "0+",
          "about": "The token HMAC.renewer principal name" },
  
      ]}
    ]},
    { "name": "RenewersThrottleTimeMs", "type": "[]DescribedDelegationTokenRenewerint32", "versions": "0+",
        "about": "ThoseThe whoduration arein ablemilliseconds tofor renewwhich thisthe tokenrequest beforewas it expires.", "fields": [
        { "name": "PrincipalType", "type": "string", "versions": "0+",
          "about": "The renewer principal type" },
        { "name": "PrincipalName", "type": "string", "versions": "0+",
          "about": "The renewer principal name" }
      ]}
    ]},
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",throttled due to a quota violation, or zero if the request did not violate any quota." }
  ]
}

AdminClient {
    //create delegation token with supplied options
	public abstract CreateDelegationTokenResult createDelegationToken(CreateDelegationTokenOptions options)
}

public class CreateDelegationTokenOptions extends AbstractOptions<CreateDelegationTokenOptions> {
    private long maxLifeTimeMs = -1;
    private List<KafkaPrincipal> renewers =  new LinkedList<>();
    private Optional<KafkaPrincipal> owner = Optional.empty(); // New this.owner = Optional.of(owner);
        return this;
    }

    public CreateDelegationTokenOptions renewers(List<KafkaPrincipal> renewers) {
        this.renewers = renewers;
        return this;
    }

    public Optional<KafkaPrincipal> owner() {
        return owner;
    }

    public List<KafkaPrincipal> renewers() {
        return renewers;
    }

    public CreateDelegationTokenOptions ownermaxlifeTimeMs(KafkaPrincipallong ownermaxLifeTimeMs) {  // New
        this.ownermaxLifeTimeMs = Optional.of(owner)maxLifeTimeMs;
        return this;
    }

    public CreateDelegationTokenOptionslong renewersmaxlifeTimeMs(List<KafkaPrincipal> renewers) {
        this.renewers = renewers;
        return thismaxLifeTimeMs;
    }

    public Optional<KafkaPrincipal> owner()}

public class TokenInformation {

    private KafkaPrincipal owner;
    private KafkaPrincipal tokenRequester; // New
    private Collection<KafkaPrincipal> renewers;
    private long issueTimestamp;
    private long maxTimestamp;
    private long expiryTimestamp;
    private String tokenId;
    ....
    ....
    ....
}