You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 3 Next »

Status

Current state: Under Discussion

Discussion thread: here

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Since Kafka transactions protocol is improved by KIP-447: Producer scalability for exactly once semantics and Kafka Streams has already onbarded eos-v2 by KIP-732: Deprecate eos-alpha and replace eos-beta with eos-v2 we should also drop support for legacy transactions protocol (without Consumer Group generation id and member id) in Kafka Producer.

Present API informs developers that they should avoid legacy protocol by marking void sendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata> offsets, String consumerGroupId) depracated, but still developers can easily use a non depreacated method void sendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata> offsets, ConsumerGroupMetadata groupMetadata) and pass there ConsumerGroupMetadata instance created by hand using non deprecated constructor.

This situation is also quite confusing for migrating the legacy applications since the easiest way to make a code not use any deprecated API is to just wrap a string consumer group id to ConsumerGroupMetadata and pass it to the void sendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata> offsets, ConsumerGroupMetadata groupMetadata) method without generation id and member id.

Users should not create instances of `ConsumerGroupMetadata`, and this class should not have any public constructor.

Public Interfaces

  • org/apache/kafka/clients/consumer/ConsumerGroupMetadata

Proposed Changes

ConsumerGroupMetadata class should be an interface, we should provide a new internal class for interface implementation and use java default scope for class visibility.

ConsumerGroupMetadata
public interface ConsumerGroupMetadata {

    String groupId();

    int generationId();

    String memberId();

    Optional<String> groupInstanceId();
}


DefaultConsumerGroupMetadata
class DefaultConsumerGroupMetadata implements ConsumerGroupMetadata {
    private final String groupId;
    private final int generationId;
    private final String memberId;
    private final Optional<String> groupInstanceId;

    DefaultConsumerGroupMetadata(String groupId,
                                        int generationId,
                                        String memberId,
                                        Optional<String> groupInstanceId) {
        this.groupId = Objects.requireNonNull(groupId, "group.id can't be null");
        this.generationId = generationId;
        this.memberId = Objects.requireNonNull(memberId, "member.id can't be null");
        this.groupInstanceId = Objects.requireNonNull(groupInstanceId, "group.instance.id can't be null");
    }

    public String groupId() {
        return groupId;
    }

    public int generationId() {
        return generationId;
    }

    public String memberId() {
        return memberId;
    }

    public Optional<String> groupInstanceId() {
        return groupInstanceId;
    }

    @Override
    public String toString() {
        return String.format("GroupMetadata(groupId = %s, generationId = %d, memberId = %s, groupInstanceId = %s)",
            groupId,
            generationId,
            memberId,
            groupInstanceId.orElse(""));
    }

    @Override
    public boolean equals(final Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        final DefaultConsumerGroupMetadata that = (DefaultConsumerGroupMetadata) o;
        return generationId == that.generationId &&
            Objects.equals(groupId, that.groupId) &&
            Objects.equals(memberId, that.memberId) &&
            Objects.equals(groupInstanceId, that.groupInstanceId);
    }

    @Override
    public int hashCode() {
        return Objects.hash(groupId, generationId, memberId, groupInstanceId);
    }

}


Compatibility, Deprecation, and Migration Plan

  • What impact (if any) will there be on existing users?
  • If we are changing behavior how will we phase out the older behavior?
  • If we need special migration tools, describe them here.
  • When will we remove the existing behavior?

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.

  • No labels