turcsanyip commented on code in PR #10467: URL: https://github.com/apache/nifi/pull/10467#discussion_r2821756986
########## nifi-extension-bundles/nifi-couchbase-bundle/nifi-couchbase-services/src/main/java/org/apache/nifi/services/couchbase/CouchbaseMapCacheClient.java: ########## @@ -0,0 +1,208 @@ +/* + * 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.nifi.services.couchbase; + +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.distributed.cache.client.AtomicCacheEntry; +import org.apache.nifi.distributed.cache.client.AtomicDistributedMapCacheClient; +import org.apache.nifi.distributed.cache.client.Deserializer; +import org.apache.nifi.distributed.cache.client.Serializer; +import org.apache.nifi.services.couchbase.exception.CouchbaseCasMismatchException; +import org.apache.nifi.services.couchbase.exception.CouchbaseDocExistsException; +import org.apache.nifi.services.couchbase.exception.CouchbaseDocNotFoundException; +import org.apache.nifi.services.couchbase.exception.CouchbaseException; +import org.apache.nifi.services.couchbase.utils.CouchbaseGetResult; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Optional; + +@Tags({"distributed", "cache", "map", "cluster", "couchbase"}) +@CapabilityDescription("Provides the ability to communicate with a Couchbase Server cluster as a DistributedMapCacheServer." + + " This can be used in order to share a Map between nodes in a NiFi cluster." + + " Couchbase Server cluster can provide a high available and persistent cache storage.") +public class CouchbaseMapCacheClient extends AbstractCouchbaseService implements AtomicDistributedMapCacheClient<Long> { + + private static final List<PropertyDescriptor> PROPERTIES = List.of( + COUCHBASE_CONNECTION_SERVICE, + BUCKET_NAME, + SCOPE_NAME, + COLLECTION_NAME + ); + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTIES; + } + + @Override + public <K, V> AtomicCacheEntry<K, V, Long> fetch(K key, Serializer<K> keySerializer, Deserializer<V> valueDeserializer) throws IOException { + final String documentId = serializeDocumentKey(key, keySerializer); + try { + final CouchbaseGetResult result = couchbaseClient.getDocument(documentId); + return new AtomicCacheEntry<>(key, deserializeDocument(valueDeserializer, result.resultContent()), result.cas()); + } catch (CouchbaseDocNotFoundException e) { + return null; + } catch (CouchbaseException e) { + throw new IOException("Failed to fetch cache entry [%s] from Couchbase".formatted(documentId), e); + } + } + + @Override + public <K, V> boolean replace(AtomicCacheEntry<K, V, Long> entry, Serializer<K> keySerializer, Serializer<V> valueSerializer) throws IOException { + final String documentId = serializeDocumentKey(entry.getKey(), keySerializer); + final byte[] document = serializeDocument(entry.getValue(), valueSerializer); + final Optional<Long> revision = entry.getRevision(); + + if (revision.isEmpty()) { + try { + couchbaseClient.insertDocument(documentId, document); + return true; + } catch (CouchbaseDocExistsException e) { + return false; + } catch (CouchbaseException e) { + throw new IOException("Failed to insert cache entry [%s] into Couchbase".formatted(documentId), e); + } + } + + try { + final long casValue = revision.get(); + couchbaseClient.replaceDocument(documentId, document, casValue); + return true; + } catch (CouchbaseDocNotFoundException | CouchbaseCasMismatchException e) { + return false; + } catch (CouchbaseException e) { + throw new IOException("Failed to replace cache entry [%s] in Couchbase".formatted(documentId), e); + } + } + + @Override + public <K, V> boolean putIfAbsent(K key, V value, Serializer<K> keySerializer, Serializer<V> valueSerializer) throws IOException { + final String documentId = serializeDocumentKey(key, keySerializer); + final byte[] document = serializeDocument(value, valueSerializer); + + try { + couchbaseClient.insertDocument(documentId, document); + return true; + } catch (CouchbaseDocExistsException e) { + return false; + } catch (CouchbaseException e) { + throw new IOException("Failed to insert cache entry [%s] into Couchbase".formatted(documentId), e); + } + } + + @Override + public <K, V> V getAndPutIfAbsent(K key, V value, Serializer<K> keySerializer, Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) throws IOException { + final V document = get(key, keySerializer, valueDeserializer); + if (document != null) { + return document; + } + + boolean putResult = putIfAbsent(key, value, keySerializer, valueSerializer); + if (!putResult) { + getAndPutIfAbsent(key, value, keySerializer, valueSerializer, valueDeserializer); + } + return null; Review Comment: The result of the recursive call must be returned. Otherwise it will fall back to `null`, even if the recursive call successfully finds the cached item. ```suggestion boolean putResult = putIfAbsent(key, value, keySerializer, valueSerializer); if (!putResult) { return getAndPutIfAbsent(key, value, keySerializer, valueSerializer, valueDeserializer); } return null; ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
