KAFKA-18319: Add task assignor interfaces #18270
Conversation
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.kafka.coordinator.group.taskassignor; |
There was a problem hiding this comment.
We could consider moving this to the streams package.
There was a problem hiding this comment.
It makes sense to put all the streams related classes in a streams package. Also, the public API will have to move to the group-coordinator-api module at some point.
| * @param clientTags The client tags for a rack-aware assignment. | ||
| * @param taskOffsets The last received cumulative task offsets of assigned tasks or dormant tasks. | ||
| */ | ||
| public record AssignmentMemberSpec(Optional<String> instanceId, |
There was a problem hiding this comment.
Compared to the feature branch, I switched to Java 17 records, which compresses the code a bit.
| /** | ||
| * The group metadata specifications required to compute the target assignment. | ||
| */ | ||
| public interface GroupSpec { |
There was a problem hiding this comment.
The consumer groups introduce these interfaces to establish a clear separation between the group coordinator code and the pluggable assignors, which may live outside the group coordinator code. We haven't included pluggable assignors in KIP-1071, but I think it still makes sense to keep these interfaces for having a clean interface for people to code against. This will pay off, if we plan on making the task assignors pluggable later .
| /** | ||
| * Mock implementation of {@link TaskAssignor} that assigns tasks to members in a round-robin fashion, with a bit of stickiness. | ||
| */ | ||
| public class MockAssignor implements TaskAssignor { |
There was a problem hiding this comment.
I included the mock assignor, since it might make sense to use it, if we want to exclude the complexities of the sticky assignor and use a dead-simple assignor instead. We may or may not remove it from the code before 4.1.
| throw new TaskAssignorException( | ||
| "Task " + taskId + " of subtopology " + subtopologyId + " is assigned to multiple members."); |
There was a problem hiding this comment.
nit: This condition isn't covered in the test.
| if (m == null) { | ||
| throw new TaskAssignorException("No member available to assign task " + i + " of subtopology " + subtopologyId); |
There was a problem hiding this comment.
nit: This condition isn't covered in the test
| /** | ||
| * @return Any configurations passed to the assignor. | ||
| */ | ||
| Map<String, String> assignmentConfigs(); |
There was a problem hiding this comment.
nit: assignmentConfigs() isn't covered in the test.
There was a problem hiding this comment.
This is only an interface. The mockassingor does not use assignment confics, so it's not used in the test. This will be tested when the interface gets a production implementation (sticky assignor).
| * @param subtopologyId String identifying the subtopology. | ||
| * @return true if the subtopology is stateful, false otherwise. | ||
| */ | ||
| boolean isStateful(String subtopologyId); |
There was a problem hiding this comment.
nit: isStateful isn't covered in the tests
There was a problem hiding this comment.
This is only an interface. The mockassingor does not distinguish between stateless and stateful tasks, so it's not used in the test. This will be tested when the interface gets a production implementation.
Introduces interfaces for defining task assignors. Task assignors are pure functions, mapping the state of the group and a topology to a target assignment. We include a mock assignor, which we will be able to use when testing / benchmarking without the complexities of the sticky task assignor and the high-availability task assignor. We may remove the mock assignor in before the streams rebalance protocol goes GA. The consumer groups introduce these interfaces to establish a clear separation between the group coordinator code and the pluggable assignors, which may live outside the group coordinator code. We have removed pluggable assignors in KIP-1071, but I think it still makes sense to keep these interfaces for having a clean interface for people to code against. This will pay off, if we plan on making the task assignors pluggable later. Reviewers: Bill Bejeck <[email protected]>, David Jacot <[email protected]>
Introduces interfaces for defining task assignors. Task assignors are pure functions, mapping the state of the group and a topology to a target assignment. We include a mock assignor, which we will be able to use when testing / benchmarking without the complexities of the sticky task assignor and the high-availability task assignor. We may remove the mock assignor in before the streams rebalance protocol goes GA. The consumer groups introduce these interfaces to establish a clear separation between the group coordinator code and the pluggable assignors, which may live outside the group coordinator code. We have removed pluggable assignors in KIP-1071, but I think it still makes sense to keep these interfaces for having a clean interface for people to code against. This will pay off, if we plan on making the task assignors pluggable later. Reviewers: Bill Bejeck <[email protected]>, David Jacot <[email protected]>
Introduces interfaces for defining task assignors. Task assignors are pure functions, mapping the state of the group and a topology to a target assignment. We include a mock assignor, which we will be able to use when testing / benchmarking without the complexities of the sticky task assignor and the high-availability task assignor. We may remove the mock assignor in before the streams rebalance protocol goes GA.
The consumer groups introduce these interfaces to establish a clear separation between the group coordinator code and the pluggable assignors, which may live outside the group coordinator code. We have removed pluggable assignors in KIP-1071, but I think it still makes sense to keep these interfaces for having a clean interface for people to code against. This will pay off, if we plan on making the task assignors pluggable later.
Committer Checklist (excluded from commit message)