Skip to content

Commit 3054383

Browse files
KAFKA-20292 [1/N]: Prepare to split TargetAssignmentBuilders (apache#22485)
To accommodate asynchronous assignments, such as those from client-side assignors and assignors offloaded to background threads, we want to split the TargetAssignmentBuilders into two: one builder for building the target assignment and another for building the target assignment records. Client-side assignors will only use the second builder. Both builders require an up-to-date view of group members at the time they are run. In the non-offloaded case, this is the same view. However, the view needs to include the unwritten member operations from the ongoing heartbeat request. Currently the operations are applied within the TargetAssignmentBuilders. To avoid duplicating the logic once the TargetAssignmentBuilders are split, we would like to lift it out and pass the TargetAssignmentBuilders the updated view of members and assignments. Add an OverlayMap class, to be used to provide the updated views of members and assignments. Reviewers: Dongnuo Lyu <dlyu@confluent.io>, David Jacot <david.jacot@gmail.com>
1 parent cc7e9d2 commit 3054383

2 files changed

Lines changed: 552 additions & 0 deletions

File tree

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.kafka.coordinator.group.util;
18+
19+
import java.util.AbstractMap;
20+
import java.util.AbstractSet;
21+
import java.util.HashMap;
22+
import java.util.HashSet;
23+
import java.util.Iterator;
24+
import java.util.Map;
25+
import java.util.NoSuchElementException;
26+
import java.util.Objects;
27+
import java.util.Set;
28+
29+
/**
30+
* A map which wraps an underlying base map and accepts incremental updates
31+
* that overlay on top of it. This class expects the underlying base map to
32+
* be immutable.
33+
*
34+
* <p>Null values are not supported.
35+
*
36+
* @param <K> The key type.
37+
* @param <V> The value type.
38+
*/
39+
public class OverlayMap<K, V> extends AbstractMap<K, V> {
40+
private final Map<K, V> base;
41+
42+
/** Entries whose keys are not in the base. */
43+
private final Map<K, V> additions = new HashMap<>();
44+
45+
/** Entries whose keys are in the base, with a value that supersedes the base value. */
46+
private final Map<K, V> replacements = new HashMap<>();
47+
48+
/** Keys that are in the base but have been removed. */
49+
private final Set<Object> removals = new HashSet<>();
50+
51+
private Set<Entry<K, V>> entrySet;
52+
53+
public OverlayMap(Map<K, V> base) {
54+
this.base = Objects.requireNonNull(base);
55+
}
56+
57+
@Override
58+
public int size() {
59+
return base.size() + additions.size() - removals.size();
60+
}
61+
62+
@Override
63+
public boolean isEmpty() {
64+
return additions.isEmpty() && removals.size() == base.size();
65+
}
66+
67+
@Override
68+
public boolean containsKey(Object key) {
69+
if (additions.containsKey(key)) return true;
70+
if (replacements.containsKey(key)) return true;
71+
if (removals.contains(key)) return false;
72+
return base.containsKey(key);
73+
}
74+
75+
@Override
76+
public V get(Object key) {
77+
if (additions.containsKey(key)) return additions.get(key);
78+
if (replacements.containsKey(key)) return replacements.get(key);
79+
if (removals.contains(key)) return null;
80+
return base.get(key);
81+
}
82+
83+
@Override
84+
public V put(K key, V value) {
85+
Objects.requireNonNull(value);
86+
87+
if (additions.containsKey(key)) {
88+
return additions.put(key, value);
89+
}
90+
if (replacements.containsKey(key)) {
91+
return replacements.put(key, value);
92+
}
93+
if (removals.remove(key)) {
94+
replacements.put(key, value);
95+
return null;
96+
}
97+
if (base.containsKey(key)) {
98+
replacements.put(key, value);
99+
return base.get(key);
100+
}
101+
additions.put(key, value);
102+
return null;
103+
}
104+
105+
@Override
106+
public V remove(Object key) {
107+
if (additions.containsKey(key)) {
108+
return additions.remove(key);
109+
}
110+
if (replacements.containsKey(key)) {
111+
V prev = replacements.remove(key);
112+
removals.add(key);
113+
return prev;
114+
}
115+
if (removals.contains(key)) {
116+
return null;
117+
}
118+
if (base.containsKey(key)) {
119+
removals.add(key);
120+
return base.get(key);
121+
}
122+
return null;
123+
}
124+
125+
@Override
126+
public void clear() {
127+
additions.clear();
128+
replacements.clear();
129+
removals.clear();
130+
removals.addAll(base.keySet());
131+
}
132+
133+
@Override
134+
public Set<Entry<K, V>> entrySet() {
135+
if (entrySet != null) return entrySet;
136+
137+
entrySet = new AbstractSet<>() {
138+
@Override
139+
public Iterator<Entry<K, V>> iterator() {
140+
return new Iterator<>() {
141+
private final Iterator<Entry<K, V>> baseIterator = base.entrySet().iterator();
142+
private final Iterator<Entry<K, V>> additionsIterator = additions.entrySet().iterator();
143+
private Entry<K, V> next = null;
144+
145+
@Override
146+
public boolean hasNext() {
147+
if (next != null) return true;
148+
while (baseIterator.hasNext()) {
149+
Entry<K, V> entry = baseIterator.next();
150+
if (replacements.containsKey(entry.getKey())) {
151+
next = Map.entry(entry.getKey(), replacements.get(entry.getKey()));
152+
} else {
153+
next = entry;
154+
}
155+
if (removals.contains(entry.getKey())) continue;
156+
return true;
157+
}
158+
if (additionsIterator.hasNext()) {
159+
next = additionsIterator.next();
160+
return true;
161+
}
162+
next = null;
163+
return false;
164+
}
165+
166+
@Override
167+
public Entry<K, V> next() {
168+
if (!hasNext()) throw new NoSuchElementException();
169+
Entry<K, V> result = next;
170+
next = null;
171+
return result;
172+
}
173+
};
174+
}
175+
176+
@Override
177+
public int size() {
178+
return OverlayMap.this.size();
179+
}
180+
};
181+
return entrySet;
182+
}
183+
184+
@Override
185+
public String toString() {
186+
return "OverlayMap(" +
187+
"base=" + base +
188+
", additions=" + additions +
189+
", replacements=" + replacements +
190+
", removals=" + removals +
191+
')';
192+
}
193+
}

0 commit comments

Comments
 (0)