Skip to content

Commit 9d4702d

Browse files
wernerdvwernerdv
committed
IGNITE-28259 Remove IgniteDhtPartitionHistorySuppliersMap
1 parent 8d1c533 commit 9d4702d

7 files changed

Lines changed: 34 additions & 142 deletions

File tree

‎modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -191,8 +191,6 @@
191191
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtDemandedPartitionsMapSerializer;
192192
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionCountersMap;
193193
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionCountersMapSerializer;
194-
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionHistorySuppliersMap;
195-
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionHistorySuppliersMapSerializer;
196194
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionsToReloadMap;
197195
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionsToReloadMapSerializer;
198196
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.latch.LatchAckMessage;
@@ -553,8 +551,6 @@ public GridIoMessageFactory(Marshaller marsh, ClassLoader clsLdr) {
553551
factory.register(507, IgniteDhtPartitionCountersMap::new,
554552
new IgniteDhtPartitionCountersMapSerializer());
555553
factory.register(508, GroupPartitionIdPair::new, new GroupPartitionIdPairSerializer());
556-
factory.register(510, IgniteDhtPartitionHistorySuppliersMap::new,
557-
new IgniteDhtPartitionHistorySuppliersMapSerializer());
558554
factory.register(513, IgniteDhtPartitionsToReloadMap::new,
559555
new IgniteDhtPartitionsToReloadMapSerializer());
560556
factory.register(517, GridPartitionStateMap::new, new GridPartitionStateMapSerializer());

‎modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@
8787
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
8888
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
8989
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleRequest;
90-
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionHistorySuppliersMap;
90+
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GroupPartitionIdPair;
9191
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionsToReloadMap;
9292
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.PartitionsExchangeAware;
9393
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.RebalanceReassignExchangeTask;
@@ -1349,7 +1349,7 @@ private void sendAllPartitions(
13491349
public GridDhtPartitionsFullMessage createPartitionsFullMessage(
13501350
@Nullable final GridDhtPartitionExchangeId exchId,
13511351
@Nullable GridCacheVersion lastVer,
1352-
@Nullable IgniteDhtPartitionHistorySuppliersMap partHistSuppliers,
1352+
@Nullable Map<UUID, Map<GroupPartitionIdPair, Long>> partHistSuppliers,
13531353
@Nullable IgniteDhtPartitionsToReloadMap partsToReload
13541354
) {
13551355
Collection<CacheGroupContext> grps = cctx.cache().cacheGroups();
@@ -1370,7 +1370,7 @@ public GridDhtPartitionsFullMessage createPartitionsFullMessage(
13701370
public GridDhtPartitionsFullMessage createPartitionsFullMessage(
13711371
@Nullable final GridDhtPartitionExchangeId exchId,
13721372
@Nullable GridCacheVersion lastVer,
1373-
@Nullable IgniteDhtPartitionHistorySuppliersMap partHistSuppliers,
1373+
@Nullable Map<UUID, Map<GroupPartitionIdPair, Long>> partHistSuppliers,
13741374
@Nullable IgniteDhtPartitionsToReloadMap partsToReload,
13751375
Collection<CacheGroupContext> grps
13761376
) {

‎modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java‎

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -324,7 +324,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
324324

325325
/** */
326326
@GridToStringExclude
327-
private final IgniteDhtPartitionHistorySuppliersMap partHistSuppliers = new IgniteDhtPartitionHistorySuppliersMap();
327+
private Map<UUID, Map<GroupPartitionIdPair, Long>> partHistSuppliers = new ConcurrentHashMap<>();
328328

329329
/** Set of nodes that cannot be used for wal rebalancing due to some reason. */
330330
private final Set<UUID> exclusionsFromHistoricalRebalance = ConcurrentHashMap.newKeySet();
@@ -587,7 +587,20 @@ public void affinityChangeMessage(CacheAffinityChangeMessage affChangeMsg) {
587587
* @return List of IDs of history supplier nodes or empty list if these doesn't exist.
588588
*/
589589
public List<UUID> partitionHistorySupplier(int grpId, int partId, long cntrSince) {
590-
List<UUID> histSuppliers = partHistSuppliers.getSupplier(grpId, partId, cntrSince);
590+
List<UUID> histSuppliers;
591+
592+
if (partHistSuppliers == null)
593+
histSuppliers = Collections.emptyList();
594+
else {
595+
histSuppliers = new ArrayList<>();
596+
597+
for (Map.Entry<UUID, Map<GroupPartitionIdPair, Long>> entry : partHistSuppliers.entrySet()) {
598+
Long historyCounter = entry.getValue().get(new GroupPartitionIdPair(grpId, partId));
599+
600+
if (historyCounter != null && historyCounter <= cntrSince)
601+
histSuppliers.add(entry.getKey());
602+
}
603+
}
591604

592605
histSuppliers.removeIf(exclusionsFromHistoricalRebalance::contains);
593606

@@ -2436,7 +2449,7 @@ private String exchangeTimingsLogMessage(String header, List<String> timings) {
24362449
// Create and destroy caches and cache proxies.
24372450
cctx.cache().onExchangeDone(this, err);
24382451

2439-
Map<GroupPartitionIdPair, Long> locReserved = partHistSuppliers.getReservations(cctx.localNodeId());
2452+
Map<GroupPartitionIdPair, Long> locReserved = partHistSuppliers.get(cctx.localNodeId());
24402453

24412454
if (locReserved != null) {
24422455
boolean success = cctx.database().reserveHistoryForPreloading(locReserved);
@@ -3537,7 +3550,9 @@ private void findCounterForReservation(
35373550
break;
35383551

35393552
if (preferWalRebalance || maxOwnerCntr - ceilingMinReserved < ownerSize) {
3540-
partHistSuppliers.put(ownerId, grpId, p, ceilingMinReserved);
3553+
Map<GroupPartitionIdPair, Long> nodeMap = partHistSuppliers.computeIfAbsent(ownerId, k -> new ConcurrentHashMap<>());
3554+
3555+
nodeMap.put(new GroupPartitionIdPair(grpId, p), ceilingMinReserved);
35413556

35423557
haveHistory.add(p);
35433558

@@ -4628,8 +4643,8 @@ private void updatePartitionFullMap(AffinityTopologyVersion resTopVer, GridDhtPa
46284643

46294644
assert partHistSuppliers.isEmpty();
46304645

4631-
partHistSuppliers.putAll(msg.partitionHistorySuppliers() != null ? msg.partitionHistorySuppliers() :
4632-
IgniteDhtPartitionHistorySuppliersMap.empty());
4646+
if (msg.partitionHistorySuppliers() != null)
4647+
partHistSuppliers = msg.partitionHistorySuppliers();
46334648

46344649
// Reserve at least 2 threads for system operations.
46354650
int parallelismLvl = U.availableThreadCount(cctx.kernalContext(), GridIoPolicy.SYSTEM_POOL, 2);

‎modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import java.util.Map;
2424
import java.util.Set;
2525
import java.util.UUID;
26+
import java.util.concurrent.ConcurrentHashMap;
2627
import java.util.stream.Collectors;
2728
import java.util.stream.IntStream;
2829
import org.apache.ignite.IgniteCheckedException;
@@ -77,7 +78,7 @@ public class GridDhtPartitionsFullMessage extends GridDhtPartitionsAbstractMessa
7778
@Order(3)
7879
@Compress
7980
@GridToStringInclude
80-
IgniteDhtPartitionHistorySuppliersMap partHistSuppliers;
81+
Map<UUID, Map<GroupPartitionIdPair, Long>> partHistSuppliers;
8182

8283
/** Partitions that must be cleared and re-loaded. */
8384
@Order(4)
@@ -144,7 +145,7 @@ public GridDhtPartitionsFullMessage() {
144145
public GridDhtPartitionsFullMessage(@Nullable GridDhtPartitionExchangeId id,
145146
@Nullable GridCacheVersion lastVer,
146147
@NotNull AffinityTopologyVersion topVer,
147-
@Nullable IgniteDhtPartitionHistorySuppliersMap partHistSuppliers,
148+
@Nullable Map<UUID, Map<GroupPartitionIdPair, Long>> partHistSuppliers,
148149
@Nullable IgniteDhtPartitionsToReloadMap partsToReload) {
149150
super(id, lastVer);
150151

@@ -325,7 +326,7 @@ public CachePartitionFullCountersMap partitionUpdateCounters(int grpId) {
325326
/**
326327
* @return Partitions history suppliers.
327328
*/
328-
public IgniteDhtPartitionHistorySuppliersMap partitionHistorySuppliers() {
329+
public Map<UUID, Map<GroupPartitionIdPair, Long>> partitionHistorySuppliers() {
329330
return partHistSuppliers;
330331
}
331332

@@ -445,7 +446,7 @@ public void topologyVersion(AffinityTopologyVersion topVer) {
445446
partCntrs = new IgniteDhtPartitionCountersMap();
446447

447448
if (partHistSuppliers == null)
448-
partHistSuppliers = new IgniteDhtPartitionHistorySuppliersMap();
449+
partHistSuppliers = new ConcurrentHashMap<>();
449450

450451
if (partsToReload == null)
451452
partsToReload = new IgniteDhtPartitionsToReloadMap();

‎modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteDhtPartitionHistorySuppliersMap.java‎

Lines changed: 0 additions & 119 deletions
This file was deleted.

‎modules/core/src/main/resources/META-INF/classnames.properties‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1193,7 +1193,6 @@ org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPre
11931193
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloaderAssignments
11941194
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtDemandedPartitionsMap
11951195
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionCountersMap
1196-
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionHistorySuppliersMap
11971196
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionsToReloadMap
11981197
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteHistoricalIterator
11991198
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteHistoricalIteratorException

‎modules/core/src/test/java/org/apache/ignite/internal/managers/communication/CompressedMessageTest.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,14 +21,14 @@
2121
import java.util.Map;
2222
import java.util.Set;
2323
import java.util.UUID;
24+
import java.util.concurrent.ConcurrentHashMap;
2425
import org.apache.ignite.internal.direct.DirectMessageReader;
2526
import org.apache.ignite.internal.direct.DirectMessageWriter;
2627
import org.apache.ignite.internal.direct.state.DirectMessageState;
2728
import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
2829
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
2930
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
3031
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GroupPartitionIdPair;
31-
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionHistorySuppliersMap;
3232
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtPartitionsToReloadMap;
3333
import org.apache.ignite.internal.util.typedef.internal.U;
3434
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -107,13 +107,13 @@ public void testWriteReadHugeMessage() {
107107

108108
/** */
109109
private GridDhtPartitionsFullMessage fullMessage() {
110-
IgniteDhtPartitionHistorySuppliersMap partHistSuppliers = new IgniteDhtPartitionHistorySuppliersMap();
110+
Map<UUID, Map<GroupPartitionIdPair, Long>> partHistSuppliers = new ConcurrentHashMap<>();
111111
IgniteDhtPartitionsToReloadMap partsToReload = new IgniteDhtPartitionsToReloadMap();
112112

113113
for (int i = 0; i < 500; i++) {
114114
UUID uuid = UUID.randomUUID();
115115

116-
partHistSuppliers.put(uuid, i, i + 1, i + 2);
116+
partHistSuppliers.put(uuid, Map.of(new GroupPartitionIdPair(i, i + 1), i + 2L));
117117
partsToReload.put(uuid, i, i + 1);
118118
}
119119

@@ -122,8 +122,8 @@ private GridDhtPartitionsFullMessage fullMessage() {
122122

123123
/** */
124124
private void assertEqualsFullMsg(GridDhtPartitionsFullMessage expected, GridDhtPartitionsFullMessage actual) {
125-
Map<UUID, Map<GroupPartitionIdPair, Long>> expHistSuppliers = U.field(expected.partitionHistorySuppliers(), "map");
126-
Map<UUID, Map<GroupPartitionIdPair, Long>> actHistSuppliers = U.field(actual.partitionHistorySuppliers(), "map");
125+
Map<UUID, Map<GroupPartitionIdPair, Long>> expHistSuppliers = expected.partitionHistorySuppliers();
126+
Map<UUID, Map<GroupPartitionIdPair, Long>> actHistSuppliers = actual.partitionHistorySuppliers();
127127

128128
assertEquals(expHistSuppliers.size(), actHistSuppliers.size());
129129

0 commit comments

Comments
 (0)