Skip to content

PR for #4610 - Add ref store size threshold property to initiate purge of old streams #4852

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
at055612 wants to merge 8 commits into
base: master
Choose a base branch
from
Open

PR for #4610 - Add ref store size threshold property to initiate purge of old streams #4852

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
ba309dc
Empty initial commit on gh-4610-ref-auto-purge [no ci]
at055612 Apr 3, 2025
1fc7bc9
gh-4610 Added compaction after purge, WIP on OffHeapStoreInfo
at055612 Apr 4, 2025
b32581a
gh-4610 Fix expected.yml and CS
at055612 Apr 7, 2025
a46315e
gh-4610 Fix CS and failing test
at055612 Apr 7, 2025
c1d57f9
Merge branch 'master' into gh-4610-ref-auto-purge
at055612 Apr 9, 2025
bf115cc
gh-4610 Add perf test
at055612 Apr 9, 2025
7a2d8e0
Add EnumSetSerde
at055612 Apr 10, 2025
92a3d17
Remove pid file
at055612 Apr 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 22 additions & 23 deletions stroom-app/src/test/java/stroom/importexport/TestJsonSerialisation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ Stream<DynamicTest> testDefaultValues() {
return buildRelatedResourceTests(clazz -> {
// Try and find the no args constructor if there is any.
if (!Modifier.isInterface(clazz.getModifiers())
&& !Modifier.isAbstract(clazz.getModifiers())) {
&& !Modifier.isAbstract(clazz.getModifiers())) {

final Constructor<?>[] constructors = clazz.getDeclaredConstructors();
Constructor<?> noArgsConstructor = null;
Expand Down Expand Up @@ -164,7 +164,6 @@ Stream<DynamicTest> testDefaultValues() {
}
}
});

}

/**
Expand All @@ -181,7 +180,7 @@ Stream<DynamicTest> testNoComplexMaps() {
for (final Field field : fields) {
// Don't care about static as they are not serialised.
if (Map.class.isAssignableFrom(field.getType())
&& !Modifier.isStatic(field.getModifiers())) {
&& !Modifier.isStatic(field.getModifiers())) {
final ParameterizedType parameterizedType = (ParameterizedType) field.getGenericType();
final Type keyType = parameterizedType.getActualTypeArguments()[0];
if (!(keyType instanceof Class && ((Class<?>) keyType).isEnum())) {
Expand Down Expand Up @@ -286,8 +285,8 @@ Stream<DynamicTest> testJsonAnnotations() {
return buildRelatedResourceTests(clazz -> {

if (!Modifier.isInterface(clazz.getModifiers())
&& !Modifier.isAbstract(clazz.getModifiers())
&& !clazz.isEnum()) {
&& !Modifier.isAbstract(clazz.getModifiers())
&& !clazz.isEnum()) {
final boolean hasJsonInclude = clazz.getAnnotation(JsonInclude.class) != null;
final boolean hasJsonPropertyOrder = clazz.getAnnotation(JsonPropertyOrder.class) != null;
final AtomicInteger jsonCreatorCount = new AtomicInteger(0);
Expand All @@ -313,7 +312,7 @@ Stream<DynamicTest> testJsonAnnotations() {

fieldPropNames = getAllFields(clazz).stream()
.filter(field -> field.getDeclaredAnnotation(JsonIgnore.class) == null
&& field.getDeclaredAnnotation(JsonProperty.class) != null)
&& field.getDeclaredAnnotation(JsonProperty.class) != null)
.map(field -> {
final JsonProperty jsonProperty = field.getDeclaredAnnotation(JsonProperty.class);
if (!jsonProperty.value().isEmpty()) {
Expand All @@ -331,8 +330,8 @@ Stream<DynamicTest> testJsonAnnotations() {
final JsonIgnore jsonIgnore = field.getDeclaredAnnotation(JsonIgnore.class);
final JsonProperty jsonProperty = field.getDeclaredAnnotation(JsonProperty.class);
if (jsonIgnore == null
&& jsonProperty == null
&& !Modifier.isStatic(field.getModifiers())) {
&& jsonProperty == null
&& !Modifier.isStatic(field.getModifiers())) {
String fieldName = field.getName();
fieldsWithoutAnnotations.add(fieldName);
}
Expand All @@ -349,7 +348,7 @@ Stream<DynamicTest> testJsonAnnotations() {
SoftAssertions.assertSoftly(softly -> {
softly.assertThat(constructorPropNames)
.describedAs("%s - JsonProperties defined in the constructor must have a " +
"corresponding JsonProperty on the field.", clazz.getName())
"corresponding JsonProperty on the field.", clazz.getName())
.containsExactlyInAnyOrderElementsOf(fieldPropNames);

softly.assertThat(hasJsonInclude)
Expand Down Expand Up @@ -506,16 +505,16 @@ private String normaliseFieldName(final String fieldName) {

private boolean isGetter(final Method method) {
return !Modifier.isAbstract(method.getModifiers()) &&
Modifier.isPublic(method.getModifiers()) &&
method.getParameterCount() == 0 &&
(method.getName().startsWith("is") || method.getName().startsWith("get"));
Modifier.isPublic(method.getModifiers()) &&
method.getParameterCount() == 0 &&
(method.getName().startsWith("is") || method.getName().startsWith("get"));
}

private boolean isSetter(final Method method) {
return !Modifier.isAbstract(method.getModifiers()) &&
Modifier.isPublic(method.getModifiers()) &&
method.getParameterCount() == 1 &&
method.getName().startsWith("set");
Modifier.isPublic(method.getModifiers()) &&
method.getParameterCount() == 1 &&
method.getName().startsWith("set");
}

private String convertMethodToFieldName(final String methodName) {
Expand Down Expand Up @@ -589,7 +588,7 @@ private static void addClass(final Set<Class<?>> stroomClasses, final Class<?> c
}

} else if (clazz.getName().startsWith(PACKAGE_START)
&& !clazz.getName().contains("StroomDuration")) { // Non POJO
&& !clazz.getName().contains("StroomDuration")) { // Non POJO
// IF the class references sub classes then include those too.
final JsonSubTypes jsonSubTypes = clazz.getAnnotation(JsonSubTypes.class);
if (jsonSubTypes != null) {
Expand Down Expand Up @@ -667,16 +666,16 @@ private List<Class<?>> getSharedClasses() {
// LOGGER.info("{}", routeClassInfo.getName());
final String name = routeClassInfo.getName();
if (name.contains(".shared.") &&
!name.contains("hadoopcommonshaded") &&
!name.contains("Util") &&
!name.contains("$") &&
!name.contains("_")) {
!name.contains("hadoopcommonshaded") &&
!name.contains("Util") &&
!name.contains("$") &&
!name.contains("_")) {
try {
final Class<?> clazz = routeClassInfo.loadClass();
if (!Modifier.isInterface(clazz.getModifiers()) &&
!Modifier.isAbstract(clazz.getModifiers()) &&
!RestResource.class.isAssignableFrom(clazz) &&
!DirectRestService.class.isAssignableFrom(clazz)) {
!Modifier.isAbstract(clazz.getModifiers()) &&
!RestResource.class.isAssignableFrom(clazz) &&
!DirectRestService.class.isAssignableFrom(clazz)) {
stroomClasses.add(clazz);
}
} catch (final IllegalArgumentException e) {
Expand Down
2 changes: 2 additions & 0 deletions stroom-config/stroom-config-app/src/test/resources/stroom/config/app/expected.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -607,6 +607,8 @@ appConfig:
refreshAfterWrite: null
statisticsMode: "INTERNAL"
referenceData:
autoPurgeEnabled: true
compactAfterPurgeEnabled: false
effectiveStreamCache:
expireAfterAccess: null
expireAfterWrite: "PT10M"
Expand Down
23 changes: 15 additions & 8 deletions stroom-lmdb/src/main/java/stroom/lmdb/AbstractLmdbDb.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,13 +108,15 @@ public abstract class AbstractLmdbDb<K, V>
private final Serde<K> keySerde;
private final Serde<V> valueSerde;
private final String dbName;
private final Dbi<ByteBuffer> lmdbDbi;
private final DbiFlags[] dbiFlags;
private final LmdbEnv lmdbEnvironment;
private final ByteBufferPool byteBufferPool;

private final int keyBufferCapacity;
private final int valueBufferCapacity;

private final DbiProxy lmdbDbi;

/**
* @param lmdbEnvironment The LMDB {@link Env} to add this DB to.
* @param byteBufferPool A self loading pool of reusable ByteBuffers.
Expand All @@ -133,8 +135,9 @@ public AbstractLmdbDb(final LmdbEnv lmdbEnvironment,
this.keySerde = keySerde;
this.valueSerde = valueSerde;
this.dbName = dbName;
this.dbiFlags = dbiFlags;
this.lmdbEnvironment = lmdbEnvironment;
this.lmdbDbi = lmdbEnvironment.openDbi(dbName, dbiFlags);
this.lmdbDbi = openDbi();
this.byteBufferPool = byteBufferPool;

int keySerdeCapacity = keySerde.getBufferCapacity();
Expand All @@ -149,6 +152,10 @@ public AbstractLmdbDb(final LmdbEnv lmdbEnvironment,
this.valueBufferCapacity = valueSerde.getBufferCapacity();
}

private DbiProxy openDbi() {
return lmdbEnvironment.openDbi(dbName, dbiFlags);
}

private static Dbi<ByteBuffer> openDbi(final Env<ByteBuffer> env,
final String name,
final DbiFlags... dbiFlags) {
Expand All @@ -168,7 +175,7 @@ public String getDbName() {
return dbName;
}

public Dbi<ByteBuffer> getLmdbDbi() {
public DbiProxy getLmdbDbi() {
return lmdbDbi;
}

Expand Down Expand Up @@ -419,7 +426,7 @@ public <T> T streamEntriesAsBytes(final Txn<ByteBuffer> txn,
final KeyRange<ByteBuffer> keyRange,
final Function<Stream<CursorIterable.KeyVal<ByteBuffer>>, T> streamFunction) {

try (CursorIterable<ByteBuffer> cursorIterable = getLmdbDbi().iterate(txn, keyRange)) {
try (CursorIterable<ByteBuffer> cursorIterable = lmdbDbi.iterate(txn, keyRange)) {
final Stream<CursorIterable.KeyVal<ByteBuffer>> stream =
StreamSupport.stream(cursorIterable.spliterator(), false);

Expand Down Expand Up @@ -927,7 +934,7 @@ public long getEntryCount() {
public void logDatabaseContents(final Txn<ByteBuffer> txn, Consumer<String> logEntryConsumer) {
LmdbUtils.logDatabaseContents(
lmdbEnvironment,
lmdbDbi,
lmdbDbi.getDbi(),
txn,
keyBuffer -> deserializeKey(keyBuffer).toString(),
valueBuffer -> deserializeValue(valueBuffer).toString(),
Expand All @@ -950,7 +957,7 @@ public void logDatabaseContents(final Txn<ByteBuffer> txn) {
public void logDatabaseContents(Consumer<String> logEntryConsumer) {
LmdbUtils.logDatabaseContents(
lmdbEnvironment,
lmdbDbi,
lmdbDbi.getDbi(),
byteBuffer -> keySerde.deserialize(byteBuffer).toString(),
byteBuffer -> valueSerde.deserialize(byteBuffer).toString(),
logEntryConsumer);
Expand All @@ -965,7 +972,7 @@ public void logDatabaseContents() {
public void logRawDatabaseContents(final Txn<ByteBuffer> txn, Consumer<String> logEntryConsumer) {
LmdbUtils.logRawDatabaseContents(
lmdbEnvironment,
lmdbDbi,
lmdbDbi.getDbi(),
txn,
logEntryConsumer);
}
Expand All @@ -986,7 +993,7 @@ public void logRawDatabaseContents(final Txn<ByteBuffer> txn) {
public void logRawDatabaseContents(Consumer<String> logEntryConsumer) {
LmdbUtils.logRawDatabaseContents(
lmdbEnvironment,
lmdbDbi,
lmdbDbi.getDbi(),
logEntryConsumer);
}

Expand Down
Loading