Page History

...

Method signature	Description	RPC used
`AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets)`	Alter offset information for a streams group.	OffsetCommit
`AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets, AlterStreamsGroupOffsetsOptions options)`	Alter offset information for a streams group.	OffsetCommit
`DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId, Set<String> topics)`	Delete offset information for a set of topics in a streams group.	OffsetDelete
`DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId, Set<String> topics, DeleteStreamsGroupOffsetsOptions options)`	Delete offset information for a set of topics in a streams group.	OffsetDelete
`DeleteStreamsGroupResult deleteStreamsGroups(Collection<String> groupIds)`	Delete streams groups from the cluster.	DeleteGroups
`DeleteStreamsGroupResult deleteStreamsGroups(Collection<String> groupIds, DeleteStreamsGroupOptions options)`	Delete streams groups from the cluster.	DeleteGroups
`DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds)`	Describe some streams groups in the cluster.	StreamsGroupDescribe
`DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds, DescribeStreamsGroupsOptions options)`	Describe some streams groups in the cluster.	StreamsGroupDescribe
`ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs)`	List the streams group offsets available in the cluster for the specified Streams groups.	OffsetFetch
`ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs, ListStreamsGroupOffsetsOptions options)`	List the streams group offsets available in the cluster for the specified Streams groups.	OffsetFetch
`ListStreamsGroupsResult listStreamsGroups()`	List the streams groups available in the cluster.	ListGroups
`ListStreamsGroupsResult listStreamsGroups(ListStreamsGroupsOptions options)`	List the streams groups available in the cluster.	ListGroups

These correspond to existing APIs for consumer groups and some of the code will be shared. The main reason for duplicating them is the naming, as the current API uses "consumerGroup" in the name, despite the RPCs just using "group". There are some differences:

...

Code Block

language	java
linenumbers	true
collapse	true

   /**
    * Alters offsets for the specified group. In order to succeed, the group must be empty.
    *
    * <p>This is a convenience method for {@link #alterStreamsGroupOffsets(String, Map, AlterStreamsGroupOffsetsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupId The group for which to alter offsets.
    * @param offsets A map of offsets by partition.
    * @return The AlterStreamsGroupOffsetsResult.
    */
   default AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets) {
       return alterStreamsGroupOffsets(groupId, offsets, new AlterStreamsGroupOffsetsOptions());
   }
 
   /**
    * Alters offsets for the specified group. In order to succeed, the group must be empty.
    *
    * <p>This operation is not transactional so it may succeed for some partitions while fail for others.
    *
    * @param groupId The group for which to alter offsets.
    * @param offsets A map of offsets by partition. Partitions not specified in the map are ignored.
    * @param options The options to use when altering the offsets.
    * @return The AlterStreamsGroupOffsetsResult.
    */
   AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets, AlterStreamsGroupOffsetsOptions options);
 
   /**
    * Delete offsets for a set of topics in a Streams group with the default options.
    *
    * <p>This is a convenience method for {@link #deleteStreamsGroupOffsets(String, Set, DeleteStreamsGroupOffsetsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupId The group for which to delete offsets.
    * @param topics The topics.
    * @return The DeleteStreamsGroupOffsetsResult.
    */
   default DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId, Set<String> topics) {
       return deleteStreamsGroupOffsets(groupId, topics, new DeleteStreamsGroupOffsetsOptions());
   }
 
   /**
    * Delete offsets for a set of topics in a Streams group.
    *
    * @param groupId The group for which to delete offsets.
    * @param topics The topics.
    * @param options The options to use when deleting offsets in a Streams group.
    * @return The DeleteStreamsGroupOffsetsResult.
    */
   DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId,
       Set<String> topics,
       DeleteStreamsGroupOffsetsOptions options);
 
   /**
    * Delete Streams groups from the cluster with the default options.
    *
    * <p>This is a convenience method for {@link #deleteStreamsGroups(Collection<String>, DeleteStreamsGroupsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupIds The IDs of the groups to delete.
    * @return The DeleteStreamsGroupsResult.
    */
   default DeleteStreamsGroupsResult deleteStreamsGroups(Collection<String> groupIds) {
       return deleteStreamsGroups(groupIds, new DeleteStreamsGroupsOptions());
   }
 
   /**
    * Delete Streams groups from the cluster.
    *
    * @param groupIds The IDs of the groups to delete.
    * @param options The options to use when deleting a Streams group.
    * @return The DeleteStreamsGroupsResult.
    */
   DeleteStreamsGroupsResult deleteStreamsGroups(Collection<String> groupIds, DeleteStreamsGroupsOptions options);
 
   /**
    * Describe some Streams groups in the cluster, with the default options.
    *
    * <p>This is a convenience method for {@link #describeStreamsGroups(Collection, DescribeStreamsGroupsOptions)}
    * with default options. See the overload for more details.
    *
    * @param groupIds The IDs of the groups to describe.
    * @return The DescribeStreamsGroupsResult.
    */
   default DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds) {
       return describeStreamsGroups(groupIds, new DescribeStreamsGroupsOptions());
   }
 
   /**
    * Describe some Streams groups in the cluster.
    *
    * @param groupIds The IDs of the groups to describe.
    * @param options  The options to use when describing the groups.
    * @return The DescribeStreamsGroupsResult.
    */
   DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds,
                                                     DescribeStreamsGroupsOptions options);
 
   /**
    * List the Streams group offsets available in the cluster for the specified Streams groups with the default options.
    *
    * <p>This is a convenience method for {@link #listStreamsGroupOffsets(Map, ListStreamsGroupOffsetsOptions)}
    * to list offsets of all partitions for the specified Streams groups with default options.
    *
    * @param groupSpecs Map of Streams group ids to a spec that specifies the topic partitions of the group to list offsets for.
    * @return The ListStreamsGroupOffsetsResult
    */
   default ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs) {
       return listStreamsGroupOffsets(groupSpecs, new ListStreamsGroupOffsetsOptions());
   }
 
   /**
    * List the Streams group offsets available in the cluster for the specified Streams groups.
    *
    * @param groupSpecs Map of Streams group ids to a spec that specifies the topic partitions of the group to list offsets for.
    * @param options The options to use when listing the Streams group offsets.
    * @return The ListStreamsGroupOffsetsResult
    */
   ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs, ListStreamsGroupOffsetsOptions options);

AlterStreamsGroupOffsetResult

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.clients.admin;
 
   /**
 * The result *of List the Streams groups available in the cluster with the default options.
    *
    * <p>This is a convenience method for {@link #listStreamsGroups(ListStreamsGroupsOptions)} with default options.{@link Admin#alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long>), AlterStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsResult {
    /**
 See the overload for more* details.
Return a future which *
succeeds if all the *alter @returnoffsets The ListStreamsGroupsResultsucceed.
     */
   default public ListStreamsGroupsResultKafkaFuture<Void> listStreamsGroupsall() {
    }
 
  return listStreamsGroups(new ListStreamsGroupsOptions()); /**
   }
 
   /**
    * List the Streams groups available in the cluster.
    *
    * @param options The options to use when listing the Streams groups.
    * @return The ListStreamsGroupsResult.
    */
   ListStreamsGroupsResult listStreamsGroups(ListStreamsGroupsOptions options);

...

  * Return a future which can be used to check the result for a given partition.
     */
    public KafkaFuture<Void> partitionResult(final TopicPartition partition) {
    }
}

AlterStreamsGroupOffsetsOptions

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
  
/**
 * Options for the {@link Admin#alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long>), AlterStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsOptions extends AbstractOptions<AlterStreamsGroupOffsetsOptions> {
}

DeleteStreamsGroupOffsetsResult

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#alterStreamsGroupOffsetsAdmin#deleteStreamsGroupOffsets(String groupId, Map<TopicPartitionSet<String>, Long>DeleteStreamsGroupOffsetsOptions), AlterStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsResultDeleteStreamsGroupOffsetsResult {
    /**
     * Return a future which succeeds only if all the alter offsetsdeletions succeed.
     */
    public KafkaFuture<Void> all() {
    }
 
    /**
     * Return a future which can be used to check the result for a given partitiontopic.
     */
    public KafkaFuture<Void> partitionResult(final TopicPartitionString partitiontopic) {
    }
}

...

DeleteStreamsGroupOffsetsOptions

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
  
/**
 * Options for the {@link Admin#alterStreamsGroupOffsetsAdmin#deleteStreamsGroupOffsets(String groupId, Map<TopicPartitionSet<String>, Long>DeleteStreamsGroupOffsetsOptions), AlterStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsOptionsDeleteStreamsGroupOffsetsOptions extends AbstractOptions<AlterStreamsGroupOffsetsOptions>AbstractOptions<DeleteStreamsGroupOffsetsOptions> {
}

...

DeleteStreamsGroupsResult

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#deleteStreamsGroupOffsetsAdmin#deleteStreamsGroups(StringCollection<String>, Set<String>, DeleteStreamsGroupOffsetsOptionsDeleteStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupOffsetsResultDeleteStreamsGroupsResult {
    /**
     * Return a future which succeeds only if all the deletions succeed.
     */
    public KafkaFuture<Void> all() {
    }
 
    /**
     * Return a future map from group id to futures which can be used to check the resultstatus forof aindividual given topicdeletions.
     */
    public KafkaFuture<Void> partitionResult(final String topicMap<String, KafkaFuture<Void>> deletedGroups() {
    }
}

...

DeleteStreamsGroupsOptions

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
  
/**
 * Options for the {@link Admin#deleteStreamsGroupOffsetsAdmin#deleteStreamsGroups(StringCollection<String>, Set<String>, DeleteStreamsGroupOffsetsOptionsDeleteStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupOffsetsOptionsDeleteStreamsGroupsOptions extends AbstractOptions<DeleteStreamsGroupOffsetsOptions>AbstractOptions<DeleteStreamsGroupsOptions> {
}

...

DescribeStreamsGroupsResult

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#deleteStreamsGroupsAdmin#describeStreamsGroups(Collection<String>, DeleteStreamsGroupsOptionsDescribeStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupsResultDescribeStreamsGroupsResult {
    /**
     * Return a future which yields all succeedsStreamsGroupDescription onlyobjects, if all the deletionsdescribes succeed.
     */
    public KafkaFuture<Map<String, KafkaFuture<Void>StreamsGroupDescription>> all() {
    }
 
    /**
     * Return a map from group id to futures which canyield be used to check the status of individual deletionsgroup descriptions.
     */
    public Map<String, KafkaFuture<Void>>KafkaFuture<StreamsGroupDescription>> deletedGroupsdescribedGroups() {
    }
}

...

StreamsGroupDescription

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
  
/**
 * Options for the {@link Admin#deleteStreamsGroups(Collection<String>, DeleteStreamsGroupsOptions)} call A detailed description of a single streams group in the cluster.
 * <p>/
public class StreamsGroupDescription {

 * The API ofpublic thisStreamsGroupDescription(
 class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupsOptions extends AbstractOptions<DeleteStreamsGroupsOptions> {
}

DescribeStreamsGroupsResult

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#describeStreamsGroups(Collection<String>, DescribeStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeStreamsGroupsResult {
    /**
     * Return a future which yields all StreamsGroupDescription objects, if all the describes succeed.
     */
    public KafkaFuture<Map<String, StreamsGroupDescription>> all() {
    }
    final String groupId,
        final int groupEpoch,
        final int targetAssignmentEpoch,
        final int topologyEpoch,
        final Collection<StreamsGroupSubtopologyDescription> subtopologies,
        final Collection<StreamsGroupMemberDescription> members,
        final StreamsGroupState state,
        final Node coordinator,
        final Set<AclOperation> authorizedOperations);

    /**
     * ReturnThe id aof mapthe fromstreams group.
 id to futures which yield*/
 group descriptions.
  public   */
 String groupId();
   public Map<String, KafkaFuture<StreamsGroupDescription>> describedGroups() {
    }
}

StreamsGroupDescription

Code Block

language	java
linenumbers	true
collapse	true

/**
 * A detailed description of* aThe singleepoch streamsof groupthe inconsumer the clustergroup.
  */
public class StreamsGroupDescription {*/

    public StreamsGroupDescriptionint groupEpoch();
 
    /**
   final String groupId,
* The epoch of the target assignment.
  final  int groupEpoch,*/
        finalpublic int targetAssignmentEpoch,();

     /**
   final int topologyEpoch,
 * The epoch of the currently used finaltopology.
 Collection<StreamsGroupSubtopologyDescription> subtopologies,
   */
    public final Collection<StreamsGroupMemberDescription> members,int topologyEpoch();
    
    /**
  final StreamsGroupState state,
 * A list of the members of finalthe Nodestreams coordinator,group.
     */
    finalpublic Set<AclOperation>Collection<StreamsGroupMemberDescription> authorizedOperationsmembers();

    /**
     * TheA idlist of the subtopologies in the streams group.
     */
    public StringCollection<StreamsGroupSubtopologyDescription> groupIdsubtopologies();
    
    /**
     * The epoch of the consumer group streams group state, or UNKNOWN if the state is too new for us to parse.
     */
    public intStreamsGroupState groupEpochstate();
 
    /**
     * The epoch of the target assignment streams group coordinator, or null if the coordinator is not known.
     */
    public intNode targetAssignmentEpochcoordinator();

      /**
     * The epoch of the currently used topologyauthorizedOperations for this group, or null if that information is not known.
     */
    public intSet<AclOperation> topologyEpochauthorizedOperations();
    
    }

StreamsGroupMemberDescription

Code Block

language	java
linenumbers	true
collapse	true

/**
     * A detailed listdescription of thea single membersmember ofin the streams group.
 */
public class StreamsGroupMemberDescription  */{

    public Collection<StreamsGroupMemberDescription> membersStreamsGroupMemberDescription();

      /**
  final String memberId,
 * A list of the subtopologies in thefinal streamsint group.memberEpoch,
     */
    publicfinal Collection<StreamsGroupSubtopologyDescription> subtopologies();

Optional<String> instanceId,
    /**
    final * The streams group state, or UNKNOWN if the state is too new for us to parse.String rackId,
        final String clientId,
     */
   final public StreamsGroupState state();

String clientHost,
    /**
    final *int ThetopologyEpoch,
 streams group coordinator, or null if the coordinatorfinal is not known.String processId,
     */
   final public Node coordinator();
Endpoint userEndpoint,
    /**
    final *Map<String, authorizedOperationsString> forclientTags,
 this group, or null if that information isfinal notList<TaskOffset> known.taskOffsets,
     */
   final public Set<AclOperation> authorizedOperations();
}

StreamsGroupMemberDescription

Code Block

language	java
linenumbers	true
collapse	true

/**
 * A detailed description of a single member in the group.
 */
public class StreamsGroupMemberDescription {

    public StreamsGroupMemberDescription(List<TaskOffset> taskEndOffsets,
        final StreamsGroupMemberAssignment assignment,
        final Optional<StreamsGroupMemberAssignment> targetAssignment,
        final Stringboolean memberId,isClassic
        final Optional<String> instanceId, );

        final String clientId,/**
     * The id finalof Stringthe clientHost,
        final int memberEpoch,group member.
        final int topologyEpoch,*/
    public    final String processId,
        final Map<String, String> clientTags,
        final List<TaskOffset> taskOffsets, 
        final List<TaskOffset> taskEndOffsets,
        final StreamsGroupMemberAssignment assignment,
        final Optional<StreamsGroupMemberAssignment> targetAssignment,
		final boolean isClassic
	String memberId();

    /**
     * The idepoch of the group member.
     */
    public Stringint memberIdmemberEpoch();

    /**
     * The id of the instance, used for static membership, if available.
     */
    public Optional<String> instanceId(); 
    
    /**
     * The clientrack idID of the group member.
     */
    public String clientIdrackId();
    
    /**
     * The client hostid of the group member.
     */
    public String clientHostclientId();

    /**
     * The epochhost of the group member. 
     */
    public intString memberEpochclientHost();

    /**
     * The epoch of the group member. 
     */
    public int memberEpoch() {
        return memberEpoch;
    }

    /**
     * The epoch of the topology present on the client.
     */
    public int topologyEpoch();
 
    /**
     * Identity of the streams instance that may have multiple clients.
     */
    public String processId(); 

    /**
     * User-defined endpoint for Interactive Queries.
     */
    public Endpoint userEndpoint() {
        return userEndpoint;
    }

    /**
     * Used for rack-aware assignment algorithm.
     */
    public Map<String, String> clientTags();

    /**
     * Cumulative offsets for tasks.
     */
    public List<TaskOffset> taskOffsets();

    /**
     * Cumulative task changelog end offsets for tasks.
     */
    public List<TaskOffset> taskEndOffsets();

    /**
     * The current assignment.
     */
    public StreamsGroupMemberAssignment assignment();

    /**
     * The target assignment.
     */
    public Optional<StreamsGroupMemberAssignment> targetAssignment();     
    
    /**
     * The flag indicating whether a member is classic.
     */
    public boolean isClassic() {
        return isClassic;
    }

    /**
     * The cumulative offset for one task.
     */
    public static class TaskOffset {

        public TaskOffset(final String subtopology, final int partition, final long offset);

        /**
         * The subtopology identifier.
         */
        public String subtopology();

        /**
         * The partition of the task.
         */
        public int partition();

        /**
         * The cumulative offset (sum of offsets in all input partitions).
         */
        public long offset();
    }
}

...

Code Block

language	java
linenumbers	true
collapse	true

/**
 * A detailed description of a single subtopology
 */
public class StreamsGroupSubtopologyDescription {

    public StreamsGroupSubtopologyDescription(
        final String subtopology,
        final List<String> sourceTopics,
        final List<String> repartitionSinkTopics,
        final Map<String, TopicInfo> stateChangelogTopics,
        final Map<String, TopicInfo> repartitionSourceTopics);

    /**
     * String to uniquely identify the subtopology.
     */
    public String subtopology();

    /**
     * The topics the topology reads from.
     */
    public List<String> sourceTopics();

    /**
     * The repartition topics the topology writes to.
     */
    public List<String> repartitionSinkTopics();

    /**
     * The set of state changelog topics associated with this sub-topology.
     */
    public Map<String, TopicInfo> stateChangelogTopics();

    /**
     * The set of source topics that are internally created repartition topics.
     */
    public Map<String, TopicInfo> repartitionSourceTopics();

    /**
     * Information about a topic.
     */
    public static class TopicInfo {

        public TopicInfo(final int partitions, final Map<String, String> topicConfigs);

        /**
         * The number of partitions in the topic.
         */
        public int partitions();

        /**
         * Configurations of the topic.
         */
        public Map<String, String> topicConfigs();
    }

}

DescribeStreamsGroupsOptions

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
  
/**
 * Options for {@link Admin#describeStreamsGroups(Collection<String>, DescribeStreamsGroupsOptions)}.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeStreamsGroupsOptions extends AbstractOptions<DescribeStreamsGroupsOptions> {
    public DescribeStreamsGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations);
 
    public boolean includeAuthorizedOperations();
}

ListStreamsGroupOffsetsResult

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupOffsetsResult {    /**
     * The set of source topics that are internally created repartition topics.
     */
    public Map<String, TopicInfo> repartitionSourceTopics();

    /**
     * Information Returnabout a futuretopic.
 which yields all Map<String, Map<TopicPartition, Long> objects, if requests for all the groups succeed.
 */
    public static class TopicInfo {

      */
  public TopicInfo(final publicint KafkaFuture<Map<Stringpartitions, final Map<TopicPartitionMap<String, Long>>>String> all() {topicConfigs);

    }
 
    /**
     * Return a future which* yieldsThe anumber map of topic partitions to offsets forin the specified grouptopic.
     */
    public*/
 KafkaFuture<Map<TopicPartition, Long>> partitionsToOffset(String groupId) {
    }
}

ListStreamsGroupOffsetsOptions

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
  
public int partitions();

        /**
 * Options for {@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)}.
 * <p>
 * TheConfigurations APIof ofthe thistopic.
 class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupOffsetsOptions extends AbstractOptions<ListStreamsGroupOffsetsOptions> {
}

...

        public Map<String, String> topicConfigs();
    }

}

DescribeStreamsGroupsOptions

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
  
/**
 * SpecificationOptions of Streams group offsets to list using for {@link Admin#listStreamsGroupOffsetsAdmin#describeStreamsGroups(Map<StringCollection<String>, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptionsDescribeStreamsGroupsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupOffsetsSpec {
  public ListStreamsGroupOffsetsSpec();
 
  /**
   * Set the topic partitions whose offsets are to be listed for a Streams group.
   */
  ListStreamsGroupOffsetsSpec topicPartitions(Collection<TopicPartition> topicPartitions);
 
  /**
   * Returns the topic partitions whose offsets are to be listed for a Streams group.
   */
  Collection<TopicPartition> topicPartitionsDescribeStreamsGroupsOptions extends AbstractOptions<DescribeStreamsGroupsOptions> {
    public DescribeStreamsGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations);
 
    public boolean includeAuthorizedOperations();
}

...

ListStreamsGroupOffsetsResult

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#listStreamsGroups(ListStreamsGroupsOptionsAdmin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupsResultListStreamsGroupOffsetsResult {
    /**
     * ReturnsReturn a future thatwhich yields eitherall anMap<String, exceptionMap<TopicPartition, or the full set of Streams group listingsLong> objects, if requests for all the groups succeed.
     */
    public KafkaFuture<Collection<StreamsGroupListing>>KafkaFuture<Map<String, Map<TopicPartition, Long>>> all() {
    }
 
    /**
     * ReturnsReturn a future which yields justa themap validof listings.
topic partitions to offsets  */
    public KafkaFuture<Collection<StreamsGroupListing>> valid() {
    }
  
    /**
     * Returns a future which yields just the errors which occurredfor the specified group.
     */
    public KafkaFuture<Collection<Throwable>> errors(KafkaFuture<Map<TopicPartition, Long>> partitionsToOffset(String groupId) {
    }
}

...

ListStreamsGroupOffsetsOptions

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
 
import org.apache.kafka.common.StreamsGroupState;
 
/**
 * AOptions listingfor of a Streams group in the cluster{@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class StreamsGroupListingListStreamsGroupOffsetsOptions {
extends  public StreamsGroupListing(String groupId);
  public StreamsGroupListing(String groupId, Optional<StreamsGroupState> state);
 
  /**
   * The id of the Streams group.
   */
  public String groupId();
 
  /**
   * The Streams group state.
   */
  public Optional<StreamsGroupState> state();
}

...

AbstractOptions<ListStreamsGroupOffsetsOptions> {
}

ListStreamsGroupOffsetsSpec

Code Block

language	java
linenumbers	true
collapse	true

package org.apache.kafka.client.admin;
 
import org.apache.kafka.common.StreamsGroupState;
 
/**
 * Options forSpecification of Streams group offsets to list using {@link Admin#listStreamsGroups(ListStreamsGroupsOptionsAdmin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupOffsetsSpec ListStreamsGroupsOptions{
 extends AbstractOptions<ListStreamsGroupsOptions> {public ListStreamsGroupOffsetsSpec();
 
   /**
     * IfSet statesthe istopic set,partitions onlywhose groups in these states willoffsets are to be returned.listed Otherwise,for alla groupsStreams are returnedgroup.
     */
    public ListStreamsGroupsOptions inStates(Set<StreamsGroupState> statesListStreamsGroupOffsetsSpec topicPartitions(Collection<TopicPartition> topicPartitions);
 
    /**
     * ReturnReturns the listtopic ofpartitions Stateswhose thatoffsets are requestedto orbe emptylisted iffor noa states have been specifiedStreams group.
     */
    public Set<StreamsGroupState> statesCollection<TopicPartition> topicPartitions();
}

GroupType

Another case is added to the org.apache.kafka.common.GroupType enum:

Enum constant	Description
`StreamsSTREAMS("Streams")`	Streams group

StreamsGroupState

GroupState

Another case is added to theA new enum org.apache.kafka.common.StreamsGroupStateGroupType is addedenum:

ASSIGNING

Enum constant
`DEAD`
NOT_READY
`EMPTY`
`STABLE`

`RECONCILING`
`UNKNOWN`

Exceptions

The following new exception is added to the org.apache.kafka.common.errors package, corresponding to the new error codes in the Kafka protocol.

StreamsInvalidTopology - The supplied topology is invalid. Returned if the client sends a topology that does not fulfill the expected invariants.

...

StreamsInvalidTopologyEpoch - The client provided a invalid topology epoch with respect to the stream group state, for example, the topology was changed by the epoch not bumped.
StreamsTopologyFenced - When a client attempts to join with an outdated topology epoch.

Command-line tools

kafka-groups.sh

We add an option --streams to the above commad-line tool, which filter the groups to show share groups. Furthermore streams becomes one of the possible values for --group-type .

kafka-streams-groups.sh

A new tool called kafka-streams-groups.sh is added for working with streams groups. It has the following options:

...

Space shortcuts

Child pages

Versions Compared

Old Version 75

New Version 76

Key

AlterStreamsGroupOffsetResult

AlterStreamsGroupOffsetsOptions

DeleteStreamsGroupOffsetsResult

DeleteStreamsGroupOffsetsOptions

DeleteStreamsGroupsResult

DeleteStreamsGroupsOptions

DescribeStreamsGroupsResult

StreamsGroupDescription

DescribeStreamsGroupsResult

StreamsGroupDescription

StreamsGroupMemberDescription

StreamsGroupMemberDescription

DescribeStreamsGroupsOptions

ListStreamsGroupOffsetsResult

ListStreamsGroupOffsetsOptions

DescribeStreamsGroupsOptions

ListStreamsGroupOffsetsResult

ListStreamsGroupOffsetsOptions

ListStreamsGroupOffsetsSpec

StreamsGroupState

GroupState

Exceptions

Command-line tools

kafka-groups.sh

kafka-streams-groups.sh