forked from apache/geode
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
GEODE-2402: Write to the lucene region buckets using a callback argument
Adding a callback argument when writing to the file and chunk regions. The file and chunk regions now have a partition listener to route the put to the correct bucket. The reason for all of this is that in some cases, the core code can can send a message that only includes the PR id and the key. We need want the core to be able to resolve the correct bucket from just those things, which requires having the PartitionListener that uses the callback argument. Added a test of putting to the file and chunk regions during GII, which is the case where the core code sends a message that includes only the PR id and the key.
- Loading branch information
1 parent
a8c6543
commit 8ce8e43
Showing
19 changed files
with
499 additions
and
36 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
79 changes: 79 additions & 0 deletions
79
...n/java/org/apache/geode/cache/lucene/internal/partition/BucketTargetingFixedResolver.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license | ||
| * agreements. See the NOTICE file distributed with this work for additional information regarding | ||
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance with the License. You may obtain a | ||
| * copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software distributed under the License | ||
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express | ||
| * or implied. See the License for the specific language governing permissions and limitations under | ||
| * the License. | ||
| */ | ||
| package org.apache.geode.cache.lucene.internal.partition; | ||
|
|
||
| import org.apache.geode.cache.EntryOperation; | ||
| import org.apache.geode.cache.FixedPartitionAttributes; | ||
| import org.apache.geode.cache.FixedPartitionResolver; | ||
| import org.apache.geode.cache.PartitionResolver; | ||
| import org.apache.geode.internal.cache.FixedPartitionAttributesImpl; | ||
| import org.apache.geode.internal.cache.PartitionedRegion; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.concurrent.ConcurrentMap; | ||
|
|
||
| /** | ||
| * A partition resolver that expects the actual bucket id to be the callback argument of all | ||
| * operations. The partition resolver reverse engineers the fixed partition name and bucket number | ||
| * from the target partitioning. | ||
| * | ||
| * This is a bit messy, mostly because there's no good way to get the FixedPartition from the actual | ||
| * bucket id without iterating over all of the fixed partitions. | ||
| */ | ||
| public class BucketTargetingFixedResolver implements FixedPartitionResolver { | ||
|
|
||
| @Override | ||
| public Object getRoutingObject(final EntryOperation opDetails) { | ||
| int targetBucketId = (Integer) opDetails.getCallbackArgument(); | ||
| final Map.Entry<String, Integer[]> targetPartition = getFixedPartition(opDetails); | ||
|
|
||
| return targetBucketId - targetPartition.getValue()[0]; | ||
| } | ||
|
|
||
| @Override | ||
| public String getName() { | ||
| return getClass().getName(); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public String getPartitionName(final EntryOperation opDetails, | ||
| @Deprecated final Set targetPartitions) { | ||
| final Map.Entry<String, Integer[]> targetPartition = getFixedPartition(opDetails); | ||
| return targetPartition.getKey(); | ||
| } | ||
|
|
||
| protected Map.Entry<String, Integer[]> getFixedPartition(final EntryOperation opDetails) { | ||
| PartitionedRegion region = (PartitionedRegion) opDetails.getRegion(); | ||
| int targetBucketId = (Integer) opDetails.getCallbackArgument(); | ||
| Map<String, Integer[]> partitions = region.getPartitionsMap(); | ||
|
|
||
| return partitions.entrySet().stream().filter(entry -> withinPartition(entry, targetBucketId)) | ||
| .findFirst().get(); | ||
| } | ||
|
|
||
| private boolean withinPartition(Map.Entry<String, Integer[]> entry, int bucketId) { | ||
| int startingBucket = entry.getValue()[0]; | ||
| int endingBucket = startingBucket + entry.getValue()[1]; | ||
| return startingBucket <= bucketId && bucketId < endingBucket; | ||
| } | ||
| } |
Oops, something went wrong.