Visitar URL original
resolved comments and modified state transitions · apache/pulsar@e0c7cfb · GitHub
Skip to content

Commit e0c7cfb

Browse files
author
Heesung Sohn
committed
resolved comments and modified state transitions
1 parent e348be3 commit e0c7cfb

6 files changed

Lines changed: 16 additions & 16 deletions

File tree

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitState.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,9 @@ public enum ServiceUnitState {
6666
Splitting; // the service unit(e.g. bundle) is in the process of splitting.
6767

6868
private static Map<ServiceUnitState, Set<ServiceUnitState>> validTransitions = Map.of(
69-
Free, Set.of(Owned, Assigned),
69+
// (Free -> Released | Splitting) transitions are required
70+
// when the topic is compacted in the middle of transfer or split.
71+
Free, Set.of(Owned, Assigned, Released, Splitting),
7072
Owned, Set.of(Assigned, Splitting, Free),
7173
Assigned, Set.of(Owned, Released, Free),
7274
Released, Set.of(Owned, Free),

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateCompactionStrategy.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ public class ServiceUnitStateCompactionStrategy implements TopicCompactionStrate
3636
private boolean checkBrokers = true;
3737

3838
public ServiceUnitStateCompactionStrategy() {
39-
schema = JSONSchema.of(ServiceUnitStateData.class);
39+
schema = Schema.JSON(ServiceUnitStateData.class);
4040
}
4141

4242
@Override

‎pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateCompactionStrategyTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,8 +50,8 @@ public void test() throws InterruptedException {
5050
assertTrue(strategy.shouldKeepLeft(data(Free), data(Assigned, "")));
5151
assertFalse(strategy.shouldKeepLeft(data(Free), data(Owned)));
5252
assertTrue(strategy.shouldKeepLeft(data(Free), data(Owned, "")));
53-
assertTrue(strategy.shouldKeepLeft(data(Free), data(Released)));
54-
assertTrue(strategy.shouldKeepLeft(data(Free), data(Splitting)));
53+
assertFalse(strategy.shouldKeepLeft(data(Free), data(Released)));
54+
assertFalse(strategy.shouldKeepLeft(data(Free), data(Splitting)));
5555

5656
assertFalse(strategy.shouldKeepLeft(data(Assigned), data(Free)));
5757
assertTrue(strategy.shouldKeepLeft(data(Assigned), data(Assigned)));

‎pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,8 +36,8 @@ public void testTransitions() {
3636
assertFalse(ServiceUnitState.isValidTransition(Free, Free));
3737
assertTrue(ServiceUnitState.isValidTransition(Free, Assigned));
3838
assertTrue(ServiceUnitState.isValidTransition(Free, Owned));
39-
assertFalse(ServiceUnitState.isValidTransition(Free, Released));
40-
assertFalse(ServiceUnitState.isValidTransition(Free, Splitting));
39+
assertTrue(ServiceUnitState.isValidTransition(Free, Released));
40+
assertTrue(ServiceUnitState.isValidTransition(Free, Splitting));
4141

4242
assertTrue(ServiceUnitState.isValidTransition(Assigned, Free));
4343
assertFalse(ServiceUnitState.isValidTransition(Assigned, Assigned));

‎pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -60,8 +60,8 @@
6060
import org.apache.pulsar.client.api.ProducerBuilder;
6161
import org.apache.pulsar.client.api.PulsarClientException;
6262
import org.apache.pulsar.client.api.Reader;
63+
import org.apache.pulsar.client.api.Schema;
6364
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
64-
import org.apache.pulsar.client.impl.schema.JSONSchema;
6565
import org.apache.pulsar.common.policies.data.ClusterData;
6666
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
6767
import org.apache.pulsar.common.policies.data.RetentionPolicies;
@@ -78,7 +78,7 @@
7878
public class ServiceUnitStateCompactionTest extends MockedPulsarServiceBaseTest {
7979
private ScheduledExecutorService compactionScheduler;
8080
private BookKeeper bk;
81-
private JSONSchema<ServiceUnitStateData> schema;
81+
private Schema<ServiceUnitStateData> schema;
8282
private ServiceUnitStateCompactionStrategy strategy;
8383

8484
private ServiceUnitState testState0 = Free;
@@ -132,7 +132,6 @@ private ServiceUnitState nextValidState(ServiceUnitState from) {
132132
.filter(to -> to != Free && to != Splitting && isValidTransition(from, to))
133133
.collect(Collectors.toList());
134134
var state= candidates.get(RANDOM.nextInt(candidates.size()));
135-
System.out.println("test state:" + state);
136135
return state;
137136
}
138137

@@ -171,7 +170,7 @@ public void setup() throws Exception {
171170
compactionScheduler = Executors.newSingleThreadScheduledExecutor(
172171
new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build());
173172
bk = pulsar.getBookKeeperClientFactory().create(this.conf, null, null, Optional.empty(), null);
174-
schema = JSONSchema.of(ServiceUnitStateData.class);
173+
schema = Schema.JSON(ServiceUnitStateData.class);
175174
strategy = new ServiceUnitStateCompactionStrategy();
176175
strategy.checkBrokers(false);
177176

@@ -222,13 +221,12 @@ TestData generateTestData() throws PulsarAdminException, PulsarClientException {
222221
for (int j = 0; j < numMessages; j++) {
223222
int keyIndex = r.nextInt(maxKeys);
224223
String key = "key" + keyIndex;
225-
ServiceUnitStateData expectedData = expected.get(key);
226-
ServiceUnitState state = expectedData == null ? nextValidState(Assigned) :
227-
r.nextInt(2) == 0 ? nextInvalidState(expectedData.state()) :
228-
nextValidState(expectedData.state());
224+
ServiceUnitStateData prev = expected.get(key);
225+
ServiceUnitState prevState = prev == null ? Free : prev.state();
226+
ServiceUnitState state = r.nextBoolean() ? nextInvalidState(prevState) :
227+
nextValidState(prevState);
229228
ServiceUnitStateData value = new ServiceUnitStateData(state, key + ":" + j);
230229
producer.newMessage().key(key).value(value).send();
231-
ServiceUnitStateData prev = expected.get(key);
232230
if (!strategy.shouldKeepLeft(prev, value)) {
233231
expected.put(key, value);
234232
}

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,9 +179,9 @@ private void handleMessage(Message<T> msg) {
179179
cur);
180180
}
181181

182-
T prev = data.get(key);
183182
boolean update = true;
184183
if (compactionStrategy != null) {
184+
T prev = data.get(key);
185185
update = !compactionStrategy.shouldKeepLeft(prev, cur);
186186
}
187187

0 commit comments

Comments
 (0)