]> git.basschouten.com Git - openhab-addons.git/blob
5b96dec86956bb9bd9971c7d205c99e627ec785c
[openhab-addons.git] /
1 /**
2  * Copyright (c) 2010-2024 Contributors to the openHAB project
3  *
4  * See the NOTICE file(s) distributed with this work for additional
5  * information.
6  *
7  * This program and the accompanying materials are made available under the
8  * terms of the Eclipse Public License 2.0 which is available at
9  * http://www.eclipse.org/legal/epl-2.0
10  *
11  * SPDX-License-Identifier: EPL-2.0
12  */
13 package org.openhab.persistence.dynamodb.internal;
14
15 import java.lang.reflect.InvocationTargetException;
16 import java.net.URI;
17 import java.time.Duration;
18 import java.time.Instant;
19 import java.time.ZonedDateTime;
20 import java.util.Collections;
21 import java.util.Iterator;
22 import java.util.List;
23 import java.util.Locale;
24 import java.util.Map;
25 import java.util.Set;
26 import java.util.concurrent.CompletableFuture;
27 import java.util.concurrent.ConcurrentHashMap;
28 import java.util.concurrent.ExecutionException;
29 import java.util.concurrent.ExecutorService;
30 import java.util.concurrent.ThreadPoolExecutor;
31 import java.util.stream.Collectors;
32
33 import javax.measure.Unit;
34
35 import org.eclipse.jdt.annotation.NonNullByDefault;
36 import org.eclipse.jdt.annotation.Nullable;
37 import org.openhab.core.common.ThreadPoolManager;
38 import org.openhab.core.config.core.ConfigurableService;
39 import org.openhab.core.i18n.UnitProvider;
40 import org.openhab.core.items.GenericItem;
41 import org.openhab.core.items.GroupItem;
42 import org.openhab.core.items.Item;
43 import org.openhab.core.items.ItemNotFoundException;
44 import org.openhab.core.items.ItemRegistry;
45 import org.openhab.core.library.items.NumberItem;
46 import org.openhab.core.library.types.QuantityType;
47 import org.openhab.core.persistence.FilterCriteria;
48 import org.openhab.core.persistence.HistoricItem;
49 import org.openhab.core.persistence.PersistenceItemInfo;
50 import org.openhab.core.persistence.PersistenceService;
51 import org.openhab.core.persistence.QueryablePersistenceService;
52 import org.openhab.core.persistence.strategy.PersistenceStrategy;
53 import org.openhab.core.types.State;
54 import org.openhab.core.types.UnDefType;
55 import org.osgi.framework.BundleContext;
56 import org.osgi.framework.Constants;
57 import org.osgi.service.component.annotations.Activate;
58 import org.osgi.service.component.annotations.Component;
59 import org.osgi.service.component.annotations.Deactivate;
60 import org.osgi.service.component.annotations.Reference;
61 import org.reactivestreams.Subscriber;
62 import org.slf4j.Logger;
63 import org.slf4j.LoggerFactory;
64
65 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
66 import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
67 import software.amazon.awssdk.awscore.defaultsmode.DefaultsMode;
68 import software.amazon.awssdk.core.async.SdkPublisher;
69 import software.amazon.awssdk.core.client.config.ClientAsyncConfiguration;
70 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
71 import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption;
72 import software.amazon.awssdk.enhanced.dynamodb.DynamoDbAsyncTable;
73 import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedAsyncClient;
74 import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
75 import software.amazon.awssdk.enhanced.dynamodb.model.QueryEnhancedRequest;
76 import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
77 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
78 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClientBuilder;
79 import software.amazon.awssdk.services.dynamodb.model.ResourceNotFoundException;
80
81 /**
82  * This is the implementation of the DynamoDB {@link PersistenceService}. It persists item values
83  * using the <a href="https://aws.amazon.com/dynamodb/">Amazon DynamoDB</a> database. The states (
84  * {@link State}) of an {@link Item} are persisted in DynamoDB tables.
85  *
86  * The service creates tables automatically, one for numbers, and one for strings.
87  *
88  * @see AbstractDynamoDBItem#fromStateNew for details how different items are persisted
89  *
90  * @author Sami Salonen - Initial contribution
91  * @author Kai Kreuzer - Migration to 3.x
92  *
93  */
94 @NonNullByDefault
95 @Component(service = { PersistenceService.class,
96         QueryablePersistenceService.class }, configurationPid = "org.openhab.dynamodb", //
97         property = Constants.SERVICE_PID + "=org.openhab.dynamodb")
98 @ConfigurableService(category = "persistence", label = "DynamoDB Persistence Service", description_uri = DynamoDBPersistenceService.CONFIG_URI)
99 public class DynamoDBPersistenceService implements QueryablePersistenceService {
100
101     private static final int MAX_CONCURRENCY = 100;
102
103     protected static final String CONFIG_URI = "persistence:dynamodb";
104
105     private static final String DYNAMODB_THREADPOOL_NAME = "dynamodbPersistenceService";
106
107     private final ItemRegistry itemRegistry;
108     private final UnitProvider unitProvider;
109     private @Nullable DynamoDbEnhancedAsyncClient client;
110     private @Nullable DynamoDbAsyncClient lowLevelClient;
111     private static final Logger logger = LoggerFactory.getLogger(DynamoDBPersistenceService.class);
112     private boolean isProperlyConfigured;
113     private @Nullable DynamoDBConfig dbConfig;
114     private @Nullable DynamoDBTableNameResolver tableNameResolver;
115     private final ExecutorService executor = ThreadPoolManager.getPool(DYNAMODB_THREADPOOL_NAME);
116     private static final Duration TIMEOUT_API_CALL = Duration.ofSeconds(60);
117     private static final Duration TIMEOUT_API_CALL_ATTEMPT = Duration.ofSeconds(5);
118     private Map<Class<? extends DynamoDBItem<?>>, DynamoDbAsyncTable<? extends DynamoDBItem<?>>> tableCache = new ConcurrentHashMap<>(
119             2);
120
121     private @Nullable URI endpointOverride;
122
123     void overrideConfig(AwsRequestOverrideConfiguration.Builder config) {
124         config.apiCallAttemptTimeout(TIMEOUT_API_CALL_ATTEMPT).apiCallTimeout(TIMEOUT_API_CALL);
125     }
126
127     void overrideConfig(ClientOverrideConfiguration.Builder config) {
128         DynamoDBConfig localDbConfig = dbConfig;
129         config.apiCallAttemptTimeout(TIMEOUT_API_CALL_ATTEMPT).apiCallTimeout(TIMEOUT_API_CALL);
130         if (localDbConfig != null) {
131             localDbConfig.getRetryPolicy().ifPresent(config::retryPolicy);
132         }
133     }
134
135     @Activate
136     public DynamoDBPersistenceService(final @Reference ItemRegistry itemRegistry,
137             final @Reference UnitProvider unitProvider) {
138         this.itemRegistry = itemRegistry;
139         this.unitProvider = unitProvider;
140     }
141
142     /**
143      * For tests
144      */
145     DynamoDBPersistenceService(final ItemRegistry itemRegistry, final UnitProvider unitProvider,
146             @Nullable URI endpointOverride) {
147         this.itemRegistry = itemRegistry;
148         this.unitProvider = unitProvider;
149         this.endpointOverride = endpointOverride;
150     }
151
152     /**
153      * For tests
154      */
155     @Nullable
156     URI getEndpointOverride() {
157         return endpointOverride;
158     }
159
160     @Nullable
161     DynamoDbAsyncClient getLowLevelClient() {
162         return lowLevelClient;
163     }
164
165     ExecutorService getExecutor() {
166         return executor;
167     }
168
169     @Nullable
170     DynamoDBTableNameResolver getTableNameResolver() {
171         return tableNameResolver;
172     }
173
174     @Nullable
175     DynamoDBConfig getDbConfig() {
176         return dbConfig;
177     }
178
179     @Activate
180     public void activate(final @Nullable BundleContext bundleContext, final Map<String, Object> config) {
181         disconnect();
182         DynamoDBConfig localDbConfig = dbConfig = DynamoDBConfig.fromConfig(config);
183         if (localDbConfig == null) {
184             // Configuration was invalid. Abort service activation.
185             // Error is already logger in fromConfig.
186             return;
187         }
188         tableNameResolver = new DynamoDBTableNameResolver(localDbConfig.getTableRevision(), localDbConfig.getTable(),
189                 localDbConfig.getTablePrefixLegacy());
190         try {
191             if (!ensureClient()) {
192                 logger.error("Error creating dynamodb database client. Aborting service activation.");
193                 return;
194             }
195         } catch (Exception e) {
196             logger.error("Error constructing dynamodb client", e);
197             return;
198         }
199
200         isProperlyConfigured = true;
201         logger.debug("dynamodb persistence service activated");
202     }
203
204     @Deactivate
205     public void deactivate() {
206         logger.debug("dynamodb persistence service deactivated");
207         logIfManyQueuedTasks();
208         disconnect();
209     }
210
211     /**
212      * Initializes Dynamo DB client and determines schema
213      *
214      * If construction fails, error is logged and false is returned.
215      *
216      * @return whether initialization was successful.
217      */
218     private boolean ensureClient() {
219         DynamoDBConfig localDbConfig = dbConfig;
220         if (localDbConfig == null) {
221             return false;
222         }
223         if (client == null) {
224             try {
225                 synchronized (this) {
226                     if (this.client != null) {
227                         return true;
228                     }
229                     DynamoDbAsyncClientBuilder lowlevelClientBuilder = DynamoDbAsyncClient.builder()
230                             .defaultsMode(DefaultsMode.STANDARD)
231                             .credentialsProvider(StaticCredentialsProvider.create(localDbConfig.getCredentials()))
232                             .httpClient(NettyNioAsyncHttpClient.builder().maxConcurrency(MAX_CONCURRENCY).build())
233                             .asyncConfiguration(
234                                     ClientAsyncConfiguration.builder()
235                                             .advancedOption(SdkAdvancedAsyncClientOption.FUTURE_COMPLETION_EXECUTOR,
236                                                     executor)
237                                             .build())
238                             .overrideConfiguration(this::overrideConfig).region(localDbConfig.getRegion());
239                     if (endpointOverride != null) {
240                         logger.debug("DynamoDB has been overriden to {}", endpointOverride);
241                         lowlevelClientBuilder.endpointOverride(endpointOverride);
242                     }
243                     DynamoDbAsyncClient lowlevelClient = lowlevelClientBuilder.build();
244                     client = DynamoDbEnhancedAsyncClient.builder().dynamoDbClient(lowlevelClient).build();
245                     this.lowLevelClient = lowlevelClient;
246                 }
247             } catch (Exception e) {
248                 logger.error("Error constructing dynamodb client", e);
249                 return false;
250             }
251         }
252         return true;
253     }
254
255     private CompletableFuture<Boolean> resolveTableSchema() {
256         DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
257         DynamoDbAsyncClient localLowLevelClient = lowLevelClient;
258         if (localTableNameResolver == null || localLowLevelClient == null) {
259             throw new IllegalStateException("tableNameResolver or localLowLevelClient not available");
260         }
261         if (localTableNameResolver.isFullyResolved()) {
262             return CompletableFuture.completedFuture(true);
263         } else {
264             synchronized (localTableNameResolver) {
265                 if (localTableNameResolver.isFullyResolved()) {
266                     return CompletableFuture.completedFuture(true);
267                 }
268                 return localTableNameResolver.resolveSchema(localLowLevelClient,
269                         b -> b.overrideConfiguration(this::overrideConfig), executor).thenApplyAsync(resolved -> {
270                             if (resolved && localTableNameResolver.getTableSchema() == ExpectedTableSchema.LEGACY) {
271                                 logger.warn(
272                                         "Using legacy table format. Is it recommended to migrate to the new table format: specify the 'table' parameter and unset the old 'tablePrefix' parameter.");
273                             }
274                             return resolved;
275                         }, executor);
276             }
277         }
278     }
279
280     private <T extends DynamoDBItem<?>> DynamoDbAsyncTable<T> getTable(Class<T> dtoClass) {
281         DynamoDbEnhancedAsyncClient localClient = client;
282         DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
283         if (!ensureClient() || localClient == null || localTableNameResolver == null) {
284             throw new IllegalStateException("Client not ready");
285         }
286         ExpectedTableSchema expectedTableSchemaRevision = localTableNameResolver.getTableSchema();
287         String tableName = localTableNameResolver.fromClass(dtoClass);
288         final TableSchema<T> schema = getDynamoDBTableSchema(dtoClass, expectedTableSchemaRevision);
289         @SuppressWarnings("unchecked") // OK since this is the only place tableCache is populated
290         DynamoDbAsyncTable<T> table = (DynamoDbAsyncTable<T>) tableCache.computeIfAbsent(dtoClass,
291                 clz -> localClient.table(tableName, schema));
292         if (table == null) {
293             // Invariant. To make null checker happy
294             throw new IllegalStateException();
295         }
296         return table;
297     }
298
299     private static <T extends DynamoDBItem<?>> TableSchema<T> getDynamoDBTableSchema(Class<T> dtoClass,
300             ExpectedTableSchema expectedTableSchemaRevision) {
301         if (dtoClass.equals(DynamoDBBigDecimalItem.class)) {
302             @SuppressWarnings("unchecked") // OK thanks to above conditional
303             TableSchema<T> schema = (TableSchema<T>) (expectedTableSchemaRevision == ExpectedTableSchema.NEW
304                     ? DynamoDBBigDecimalItem.TABLE_SCHEMA_NEW
305                     : DynamoDBBigDecimalItem.TABLE_SCHEMA_LEGACY);
306             return schema;
307         } else if (dtoClass.equals(DynamoDBStringItem.class)) {
308             @SuppressWarnings("unchecked") // OK thanks to above conditional
309             TableSchema<T> schema = (TableSchema<T>) (expectedTableSchemaRevision == ExpectedTableSchema.NEW
310                     ? DynamoDBStringItem.TABLE_SCHEMA_NEW
311                     : DynamoDBStringItem.TABLE_SCHEMA_LEGACY);
312             return schema;
313         } else {
314             throw new IllegalStateException("Unknown DTO class. Bug");
315         }
316     }
317
318     private void disconnect() {
319         DynamoDbAsyncClient localLowLevelClient = lowLevelClient;
320         if (client == null || localLowLevelClient == null) {
321             return;
322         }
323         localLowLevelClient.close();
324         lowLevelClient = null;
325         client = null;
326         dbConfig = null;
327         tableNameResolver = null;
328         isProperlyConfigured = false;
329         tableCache.clear();
330     }
331
332     protected boolean isReadyToStore() {
333         return isProperlyConfigured && ensureClient();
334     }
335
336     @Override
337     public String getId() {
338         return "dynamodb";
339     }
340
341     @Override
342     public String getLabel(@Nullable Locale locale) {
343         return "DynamoDB";
344     }
345
346     @Override
347     public Set<PersistenceItemInfo> getItemInfo() {
348         return Collections.emptySet();
349     }
350
351     @Override
352     public Iterable<HistoricItem> query(FilterCriteria filter) {
353         logIfManyQueuedTasks();
354         Instant start = Instant.now();
355         String filterDescription = filterToString(filter);
356         logger.trace("Got a query with filter {}", filterDescription);
357         DynamoDbEnhancedAsyncClient localClient = client;
358         DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
359         if (!isProperlyConfigured) {
360             logger.debug("Configuration for dynamodb not yet loaded or broken. Returning empty query results.");
361             return Collections.emptyList();
362         }
363         if (!ensureClient() || localClient == null || localTableNameResolver == null) {
364             logger.warn("DynamoDB not connected. Returning empty query results.");
365             return Collections.emptyList();
366         }
367
368         //
369         // Resolve unclear table schema if needed
370         //
371         try {
372             Boolean resolved = resolveTableSchema().get();
373             if (!resolved) {
374                 logger.warn("Table schema not resolved, cannot query data.");
375                 return Collections.emptyList();
376             }
377         } catch (InterruptedException e) {
378             logger.warn("Table schema resolution interrupted, cannot query data");
379             return Collections.emptyList();
380         } catch (ExecutionException e) {
381             Throwable cause = e.getCause();
382             logger.warn("Table schema resolution errored, cannot query data: {} {}",
383                     cause == null ? e.getClass().getSimpleName() : cause.getClass().getSimpleName(),
384                     cause == null ? e.getMessage() : cause.getMessage());
385             return Collections.emptyList();
386         }
387         try {
388             //
389             // Proceed with query
390             //
391             String itemName = filter.getItemName();
392             if (itemName == null) {
393                 logger.warn("Item name is missing in filter {}", filter);
394                 return List.of();
395             }
396             Item item = getItemFromRegistry(itemName);
397             if (item == null) {
398                 logger.warn("Could not get item {} from registry! Returning empty query results.", itemName);
399                 return Collections.emptyList();
400             }
401             if (item instanceof GroupItem groupItem) {
402                 item = groupItem.getBaseItem();
403                 logger.debug("Item is instanceof GroupItem '{}'", itemName);
404                 if (item == null) {
405                     logger.debug("BaseItem of GroupItem is null. Ignore and give up!");
406                     return Collections.emptyList();
407                 }
408                 if (item instanceof GroupItem) {
409                     logger.debug("BaseItem of GroupItem is a GroupItem too. Ignore and give up!");
410                     return Collections.emptyList();
411                 }
412             }
413             boolean legacy = localTableNameResolver.getTableSchema() == ExpectedTableSchema.LEGACY;
414             Class<? extends DynamoDBItem<?>> dtoClass = AbstractDynamoDBItem.getDynamoItemClass(item.getClass(),
415                     legacy);
416             String tableName = localTableNameResolver.fromClass(dtoClass);
417             DynamoDbAsyncTable<? extends DynamoDBItem<?>> table = getTable(dtoClass);
418             logger.debug("Item {} (of type {}) will be tried to query using DTO class {} from table {}", itemName,
419                     item.getClass().getSimpleName(), dtoClass.getSimpleName(), tableName);
420
421             QueryEnhancedRequest queryExpression = DynamoDBQueryUtils.createQueryExpression(dtoClass,
422                     localTableNameResolver.getTableSchema(), item, filter, unitProvider);
423
424             CompletableFuture<List<DynamoDBItem<?>>> itemsFuture = new CompletableFuture<>();
425             final SdkPublisher<? extends DynamoDBItem<?>> itemPublisher = table.query(queryExpression).items();
426             Subscriber<DynamoDBItem<?>> pageSubscriber = new PageOfInterestSubscriber<>(itemsFuture,
427                     filter.getPageNumber(), filter.getPageSize());
428             itemPublisher.subscribe(pageSubscriber);
429             // NumberItem.getUnit() is expensive, we avoid calling it in the loop
430             // by fetching the unit here.
431             final Item localItem = item;
432             final Unit<?> itemUnit = localItem instanceof NumberItem ni ? ni.getUnit() : null;
433             try {
434                 @SuppressWarnings("null")
435                 List<HistoricItem> results = itemsFuture.get().stream().map(dynamoItem -> {
436                     HistoricItem historicItem = dynamoItem.asHistoricItem(localItem, itemUnit);
437                     if (historicItem == null) {
438                         logger.warn(
439                                 "Dynamo item {} serialized state '{}' cannot be converted to item {} {}. Item type changed since persistence. Ignoring",
440                                 dynamoItem.getClass().getSimpleName(), dynamoItem.getState(),
441                                 localItem.getClass().getSimpleName(), localItem.getName());
442                         return null;
443                     }
444                     logger.trace("Dynamo item {} converted to historic item: {}", localItem, historicItem);
445                     return historicItem;
446                 }).filter(value -> value != null).collect(Collectors.toList());
447                 logger.debug("Query completed in {} ms. Filter was {}",
448                         Duration.between(start, Instant.now()).toMillis(), filterDescription);
449                 return results;
450             } catch (InterruptedException e) {
451                 logger.warn("Query interrupted. Filter was {}", filterDescription);
452                 return Collections.emptyList();
453             } catch (ExecutionException e) {
454                 Throwable cause = e.getCause();
455                 if (cause instanceof ResourceNotFoundException) {
456                     logger.trace("Query failed since the DynamoDB table '{}' does not exist. Filter was {}", tableName,
457                             filterDescription);
458                 } else if (logger.isTraceEnabled()) {
459                     logger.trace("Query failed. Filter was {}", filterDescription, e);
460                 } else {
461                     logger.warn("Query failed {} {}. Filter was {}",
462                             cause == null ? e.getClass().getSimpleName() : cause.getClass().getSimpleName(),
463                             cause == null ? e.getMessage() : cause.getMessage(), filterDescription);
464                 }
465                 return Collections.emptyList();
466             }
467         } catch (Exception e) {
468             logger.error("Unexpected error with query having filter {}: {} {}. Returning empty query results.",
469                     filterDescription, e.getClass().getSimpleName(), e.getMessage());
470             return Collections.emptyList();
471         }
472     }
473
474     /**
475      * Retrieves the item for the given name from the item registry
476      *
477      * @param itemName
478      * @return item with the given name, or null if no such item exists in item registry.
479      */
480     private @Nullable Item getItemFromRegistry(String itemName) {
481         try {
482             return itemRegistry.getItem(itemName);
483         } catch (ItemNotFoundException e1) {
484             return null;
485         }
486     }
487
488     @Override
489     public List<PersistenceStrategy> getDefaultStrategies() {
490         return List.of(PersistenceStrategy.Globals.RESTORE, PersistenceStrategy.Globals.CHANGE);
491     }
492
493     @Override
494     public void store(Item item) {
495         store(item, null);
496     }
497
498     @Override
499     public void store(Item item, @Nullable String alias) {
500         // Timestamp and capture state immediately as rest of the store is asynchronous (state might change in between)
501         ZonedDateTime time = ZonedDateTime.now();
502
503         logIfManyQueuedTasks();
504         if (!(item instanceof GenericItem)) {
505             return;
506         }
507         if (item.getState() instanceof UnDefType) {
508             logger.debug("Undefined item state received. Not storing item {}.", item.getName());
509             return;
510         }
511         if (!isReadyToStore()) {
512             logger.warn("Not ready to store (config error?), not storing item {}.", item.getName());
513             return;
514         }
515         // Get Item describing the real type of data
516         // With non-group items this is same as the argument item. With Group items, this is item describing the type of
517         // state stored in the group.
518         final Item itemTemplate;
519         try {
520             itemTemplate = getEffectiveItem(item);
521         } catch (IllegalStateException e) {
522             // Exception is raised when underlying item type cannot be determined with Group item
523             // Logged already
524             return;
525         }
526
527         String effectiveName = (alias != null) ? alias : item.getName();
528
529         // We do not want to rely item.state since async context below can execute much later.
530         // We 'copy' the item for local use. copyItem also normalizes the unit with NumberItems.
531         final GenericItem copiedItem = copyItem(itemTemplate, item, effectiveName, null, unitProvider);
532
533         resolveTableSchema().thenAcceptAsync(resolved -> {
534             if (!resolved) {
535                 logger.warn("Table schema not resolved, not storing item {}.", copiedItem.getName());
536                 return;
537             }
538
539             DynamoDbEnhancedAsyncClient localClient = client;
540             DynamoDbAsyncClient localLowlevelClient = lowLevelClient;
541             DynamoDBConfig localConfig = dbConfig;
542             DynamoDBTableNameResolver localTableNameResolver = tableNameResolver;
543             if (!isProperlyConfigured || localClient == null || localLowlevelClient == null || localConfig == null
544                     || localTableNameResolver == null) {
545                 logger.warn("Not ready to store (config error?), not storing item {}.", item.getName());
546                 return;
547             }
548
549             Integer expireDays = localConfig.getExpireDays();
550
551             final DynamoDBItem<?> dto;
552             switch (localTableNameResolver.getTableSchema()) {
553                 case NEW:
554                     dto = AbstractDynamoDBItem.fromStateNew(copiedItem, time, expireDays);
555                     break;
556                 case LEGACY:
557                     dto = AbstractDynamoDBItem.fromStateLegacy(copiedItem, time);
558                     break;
559                 default:
560                     throw new IllegalStateException("Unexpected. Bug");
561             }
562             logger.trace("store() called with item {} {} '{}', which was converted to DTO {}",
563                     copiedItem.getClass().getSimpleName(), effectiveName, copiedItem.getState(), dto);
564             dto.accept(new DynamoDBItemVisitor<TableCreatingPutItem<? extends DynamoDBItem<?>>>() {
565
566                 @Override
567                 public TableCreatingPutItem<? extends DynamoDBItem<?>> visit(
568                         DynamoDBBigDecimalItem dynamoBigDecimalItem) {
569                     return new TableCreatingPutItem<>(DynamoDBPersistenceService.this, dynamoBigDecimalItem,
570                             getTable(DynamoDBBigDecimalItem.class));
571                 }
572
573                 @Override
574                 public TableCreatingPutItem<? extends DynamoDBItem<?>> visit(DynamoDBStringItem dynamoStringItem) {
575                     return new TableCreatingPutItem<>(DynamoDBPersistenceService.this, dynamoStringItem,
576                             getTable(DynamoDBStringItem.class));
577                 }
578             }).putItemAsync();
579         }, executor).exceptionally(e -> {
580             logger.error("Unexcepted error", e);
581             return null;
582         });
583     }
584
585     private Item getEffectiveItem(Item item) {
586         final Item effectiveItem;
587         if (item instanceof GroupItem groupItem) {
588             Item baseItem = groupItem.getBaseItem();
589             if (baseItem == null) {
590                 // if GroupItem:<ItemType> is not defined in
591                 // *.items using StringType
592                 logger.debug(
593                         "Cannot detect ItemType for {} because the GroupItems' base type isn't set in *.items File.",
594                         item.getName());
595                 Iterator<Item> firstGroupMemberItem = groupItem.getMembers().iterator();
596                 if (firstGroupMemberItem.hasNext()) {
597                     effectiveItem = firstGroupMemberItem.next();
598                 } else {
599                     throw new IllegalStateException("GroupItem " + item.getName()
600                             + " does not have children nor base item set, cannot determine underlying item type. Aborting!");
601                 }
602             } else {
603                 effectiveItem = baseItem;
604             }
605         } else {
606             effectiveItem = item;
607         }
608         return effectiveItem;
609     }
610
611     /**
612      * Copy item and optionally override name and state
613      *
614      * State is normalized to source item's unit with Quantity NumberItems and QuantityTypes
615      *
616      * @param itemTemplate 'template item' to be used to construct the new copy. It is also used to determine UoM unit
617      *            and get GenericItem.type
618      * @param item item that is used to acquire name and state
619      * @param nameOverride name override for the resulting copy
620      * @param stateOverride state override for the resulting copy
621      * @param unitProvider the unit provider for number with dimension
622      * @throws IllegalArgumentException when state is QuantityType and not compatible with item
623      */
624     static GenericItem copyItem(Item itemTemplate, Item item, @Nullable String nameOverride,
625             @Nullable State stateOverride, UnitProvider unitProvider) {
626         final GenericItem copiedItem;
627         try {
628             if (itemTemplate instanceof NumberItem) {
629                 copiedItem = (GenericItem) itemTemplate.getClass()
630                         .getDeclaredConstructor(String.class, String.class, UnitProvider.class)
631                         .newInstance(itemTemplate.getType(), nameOverride == null ? item.getName() : nameOverride,
632                                 unitProvider);
633             } else {
634                 copiedItem = (GenericItem) itemTemplate.getClass().getDeclaredConstructor(String.class)
635                         .newInstance(nameOverride == null ? item.getName() : nameOverride);
636             }
637
638         } catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException
639                 | NoSuchMethodException | SecurityException e) {
640             throw new IllegalArgumentException(item.toString(), e);
641         }
642         State state = stateOverride == null ? item.getState() : stateOverride;
643         if (state instanceof QuantityType<?> type && itemTemplate instanceof NumberItem numberItem) {
644             Unit<?> itemUnit = numberItem.getUnit();
645             if (itemUnit != null) {
646                 State convertedState = type.toUnit(itemUnit);
647                 if (convertedState == null) {
648                     logger.error("Unexpected unit conversion failure: {} to item unit {}", state, itemUnit);
649                     throw new IllegalArgumentException(
650                             String.format("Unexpected unit conversion failure: %s to item unit %s", state, itemUnit));
651                 }
652                 state = convertedState;
653             }
654         }
655         copiedItem.setState(state);
656         return copiedItem;
657     }
658
659     private void logIfManyQueuedTasks() {
660         if (executor instanceof ThreadPoolExecutor localExecutor) {
661             if (localExecutor.getQueue().size() >= 5) {
662                 logger.trace("executor queue size: {}, remaining space {}. Active threads {}",
663                         localExecutor.getQueue().size(), localExecutor.getQueue().remainingCapacity(),
664                         localExecutor.getActiveCount());
665             } else if (localExecutor.getQueue().size() >= 50) {
666                 logger.warn(
667                         "Many ({}) tasks queued in executor! This might be sign of bad design or bug in the addon code.",
668                         localExecutor.getQueue().size());
669             }
670         }
671     }
672
673     private String filterToString(FilterCriteria filter) {
674         return String.format(
675                 "FilterCriteria@%s(item=%s, pageNumber=%d, pageSize=%d, time=[%s, %s, %s], state=[%s, %s of %s] )",
676                 System.identityHashCode(filter), filter.getItemName(), filter.getPageNumber(), filter.getPageSize(),
677                 filter.getBeginDate(), filter.getEndDate(), filter.getOrdering(), filter.getOperator(),
678                 filter.getState(), filter.getState() == null ? "null" : filter.getState().getClass().getSimpleName());
679     }
680 }