001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.wal;
019
020import java.io.FileNotFoundException;
021import java.io.IOException;
022import java.io.UnsupportedEncodingException;
023import java.net.URLDecoder;
024import java.nio.charset.StandardCharsets;
025import java.util.ArrayList;
026import java.util.Collections;
027import java.util.Comparator;
028import java.util.List;
029import java.util.concurrent.locks.ReadWriteLock;
030import java.util.concurrent.locks.ReentrantReadWriteLock;
031import java.util.regex.Matcher;
032import java.util.regex.Pattern;
033import org.apache.hadoop.conf.Configuration;
034import org.apache.hadoop.fs.FileStatus;
035import org.apache.hadoop.fs.FileSystem;
036import org.apache.hadoop.fs.Path;
037import org.apache.hadoop.hbase.HConstants;
038import org.apache.hadoop.hbase.ServerName;
039import org.apache.hadoop.hbase.client.RegionInfo;
040import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
041import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
042import org.apache.hadoop.hbase.util.Addressing;
043import org.apache.hadoop.hbase.util.CancelableProgressable;
044import org.apache.hadoop.hbase.util.CommonFSUtils;
045import org.apache.hadoop.hbase.util.RecoverLeaseFSUtils;
046import org.apache.yetus.audience.InterfaceAudience;
047import org.apache.yetus.audience.InterfaceStability;
048import org.slf4j.Logger;
049import org.slf4j.LoggerFactory;
050
051import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
052
053/**
054 * Base class of a WAL Provider that returns a single thread safe WAL that writes to Hadoop FS. By
055 * default, this implementation picks a directory in Hadoop FS based on a combination of
056 * <ul>
057 * <li>the HBase root directory
058 * <li>HConstants.HREGION_LOGDIR_NAME
059 * <li>the given factory's factoryId (usually identifying the regionserver by host:port)
060 * </ul>
061 * It also uses the providerId to differentiate among files.
062 */
063@InterfaceAudience.Private
064@InterfaceStability.Evolving
065public abstract class AbstractFSWALProvider<T extends AbstractFSWAL<?>>
066  extends AbstractWALProvider {
067
068  private static final Logger LOG = LoggerFactory.getLogger(AbstractFSWALProvider.class);
069
070  /** Separate old log into different dir by regionserver name **/
071  public static final String SEPARATE_OLDLOGDIR = "hbase.separate.oldlogdir.by.regionserver";
072  public static final boolean DEFAULT_SEPARATE_OLDLOGDIR = false;
073
074  public interface Initializer {
075    /**
076     * A method to initialize a WAL reader.
077     * @param startPosition the start position you want to read from, -1 means start reading from
078     *                      the first WAL entry. Notice that, the first entry is not started at
079     *                      position as we have several headers, so typically you should not pass 0
080     *                      here.
081     */
082    void init(FileSystem fs, Path path, Configuration c, long startPosition) throws IOException;
083  }
084
085  protected volatile T wal;
086
087  /**
088   * We use walCreateLock to prevent wal recreation in different threads, and also prevent getWALs
089   * missing the newly created WAL, see HBASE-21503 for more details.
090   */
091  private final ReadWriteLock walCreateLock = new ReentrantReadWriteLock();
092
093  /**
094   * @param factory    factory that made us, identity used for FS layout. may not be null
095   * @param conf       may not be null
096   * @param providerId differentiate between providers from one factory, used for FS layout. may be
097   *                   null
098   */
099  @Override
100  protected void doInit(WALFactory factory, Configuration conf, String providerId)
101    throws IOException {
102    this.providerId = providerId;
103    // get log prefix
104    StringBuilder sb = new StringBuilder().append(factory.factoryId);
105    if (providerId != null) {
106      if (providerId.startsWith(WAL_FILE_NAME_DELIMITER)) {
107        sb.append(providerId);
108      } else {
109        sb.append(WAL_FILE_NAME_DELIMITER).append(providerId);
110      }
111    }
112    logPrefix = sb.toString();
113    doInit(conf);
114  }
115
116  @Override
117  protected List<WAL> getWALs0() {
118    if (wal != null) {
119      return Lists.newArrayList(wal);
120    }
121    walCreateLock.readLock().lock();
122    try {
123      if (wal == null) {
124        return Collections.emptyList();
125      } else {
126        return Lists.newArrayList(wal);
127      }
128    } finally {
129      walCreateLock.readLock().unlock();
130    }
131  }
132
133  @Override
134  protected T getWAL0(RegionInfo region) throws IOException {
135    T walCopy = wal;
136    if (walCopy != null) {
137      return walCopy;
138    }
139    walCreateLock.writeLock().lock();
140    try {
141      walCopy = wal;
142      if (walCopy != null) {
143        return walCopy;
144      }
145      walCopy = createWAL();
146      initWAL(walCopy);
147      wal = walCopy;
148      return walCopy;
149    } finally {
150      walCreateLock.writeLock().unlock();
151    }
152  }
153
154  protected abstract T createWAL() throws IOException;
155
156  protected abstract void doInit(Configuration conf) throws IOException;
157
158  @Override
159  protected void shutdown0() throws IOException {
160    T log = this.wal;
161    if (log != null) {
162      log.shutdown();
163    }
164  }
165
166  @Override
167  protected void close0() throws IOException {
168    T log = this.wal;
169    if (log != null) {
170      log.close();
171    }
172  }
173
174  /**
175   * iff the given WALFactory is using the DefaultWALProvider for meta and/or non-meta, count the
176   * number of files (rolled and active). if either of them aren't, count 0 for that provider.
177   */
178  @Override
179  protected long getNumLogFiles0() {
180    T log = this.wal;
181    return log == null ? 0 : log.getNumLogFiles();
182  }
183
184  /**
185   * iff the given WALFactory is using the DefaultWALProvider for meta and/or non-meta, count the
186   * size of files (only rolled). if either of them aren't, count 0 for that provider.
187   */
188  @Override
189  protected long getLogFileSize0() {
190    T log = this.wal;
191    return log == null ? 0 : log.getLogFileSize();
192  }
193
194  @Override
195  protected WAL createRemoteWAL(RegionInfo region, FileSystem remoteFs, Path remoteWALDir,
196    String prefix, String suffix) throws IOException {
197    // so we do not need to add this for a lot of test classes, for normal WALProvider, you should
198    // implement this method to support sync replication.
199    throw new UnsupportedOperationException();
200  }
201
202  /**
203   * returns the number of rolled WAL files.
204   */
205  public static int getNumRolledLogFiles(WAL wal) {
206    return ((AbstractFSWAL<?>) wal).getNumRolledLogFiles();
207  }
208
209  /**
210   * returns the size of rolled WAL files.
211   */
212  public static long getLogFileSize(WAL wal) {
213    return ((AbstractFSWAL<?>) wal).getLogFileSize();
214  }
215
216  /**
217   * return the current filename from the current wal.
218   */
219  public static Path getCurrentFileName(final WAL wal) {
220    return ((AbstractFSWAL<?>) wal).getCurrentFileName();
221  }
222
223  /**
224   * request a log roll, but don't actually do it.
225   */
226  static void requestLogRoll(final WAL wal) {
227    ((AbstractFSWAL<?>) wal).requestLogRoll();
228  }
229
230  // should be package private; more visible for use in AbstractFSWAL
231  public static final String WAL_FILE_NAME_DELIMITER = ".";
232  /** The hbase:meta region's WAL filename extension */
233  public static final String META_WAL_PROVIDER_ID = ".meta";
234  static final String DEFAULT_PROVIDER_ID = "default";
235
236  // Implementation details that currently leak in tests or elsewhere follow
237  /** File Extension used while splitting an WAL into regions (HBASE-2312) */
238  public static final String SPLITTING_EXT = "-splitting";
239
240  /**
241   * Pattern used to validate a WAL file name see {@link #validateWALFilename(String)} for
242   * description.
243   */
244  private static final Pattern WAL_FILE_NAME_PATTERN =
245    Pattern.compile("(.+)\\.(\\d+)(\\.[0-9A-Za-z]+)?");
246
247  /**
248   * Define for when no timestamp found.
249   */
250  private static final long NO_TIMESTAMP = -1L;
251
252  /**
253   * It returns the file create timestamp (the 'FileNum') from the file name. For name format see
254   * {@link #validateWALFilename(String)} public until remaining tests move to o.a.h.h.wal
255   * @param wal must not be null
256   * @return the file number that is part of the WAL file name
257   */
258  public static long extractFileNumFromWAL(final WAL wal) {
259    final Path walPath = ((AbstractFSWAL<?>) wal).getCurrentFileName();
260    if (walPath == null) {
261      throw new IllegalArgumentException("The WAL path couldn't be null");
262    }
263    String name = walPath.getName();
264    long timestamp = getTimestamp(name);
265    if (timestamp == NO_TIMESTAMP) {
266      throw new IllegalArgumentException(name + " is not a valid wal file name");
267    }
268    return timestamp;
269  }
270
271  /**
272   * A WAL file name is of the format: &lt;wal-name&gt;{@link #WAL_FILE_NAME_DELIMITER}
273   * &lt;file-creation-timestamp&gt;[.&lt;suffix&gt;]. provider-name is usually made up of a
274   * server-name and a provider-id
275   * @param filename name of the file to validate
276   * @return <tt>true</tt> if the filename matches an WAL, <tt>false</tt> otherwise
277   */
278  public static boolean validateWALFilename(String filename) {
279    return WAL_FILE_NAME_PATTERN.matcher(filename).matches();
280  }
281
282  /**
283   * Split a WAL filename to get a start time. WALs usually have the time we start writing to them
284   * with as part of their name, usually the suffix. Sometimes there will be an extra suffix as when
285   * it is a WAL for the meta table. For example, WALs might look like this
286   * <code>10.20.20.171%3A60020.1277499063250</code> where <code>1277499063250</code> is the
287   * timestamp. Could also be a meta WAL which adds a '.meta' suffix or a synchronous replication
288   * WAL which adds a '.syncrep' suffix. Check for these. File also may have no timestamp on it. For
289   * example the recovered.edits files are WALs but are named in ascending order. Here is an
290   * example: 0000000000000016310. Allow for this.
291   * @param name Name of the WAL file.
292   * @return Timestamp or {@link #NO_TIMESTAMP}.
293   */
294  public static long getTimestamp(String name) {
295    Matcher matcher = WAL_FILE_NAME_PATTERN.matcher(name);
296    return matcher.matches() ? Long.parseLong(matcher.group(2)) : NO_TIMESTAMP;
297  }
298
299  public static final Comparator<Path> TIMESTAMP_COMPARATOR =
300    Comparator.<Path, Long> comparing(p -> AbstractFSWALProvider.getTimestamp(p.getName()))
301      .thenComparing(Path::getName);
302
303  /**
304   * Construct the directory name for all WALs on a given server. Dir names currently look like this
305   * for WALs: <code>hbase//WALs/kalashnikov.att.net,61634,1486865297088</code>.
306   * @param serverName Server name formatted as described in {@link ServerName}
307   * @return the relative WAL directory name, e.g. <code>.logs/1.example.org,60030,12345</code> if
308   *         <code>serverName</code> passed is <code>1.example.org,60030,12345</code>
309   */
310  public static String getWALDirectoryName(final String serverName) {
311    StringBuilder dirName = new StringBuilder(HConstants.HREGION_LOGDIR_NAME);
312    dirName.append("/");
313    dirName.append(serverName);
314    return dirName.toString();
315  }
316
317  /**
318   * Construct the directory name for all old WALs on a given server. The default old WALs dir looks
319   * like: <code>hbase/oldWALs</code>. If you config hbase.separate.oldlogdir.by.regionserver to
320   * true, it looks like <code>hbase//oldWALs/kalashnikov.att.net,61634,1486865297088</code>.
321   * @param serverName Server name formatted as described in {@link ServerName}
322   * @return the relative WAL directory name
323   */
324  public static String getWALArchiveDirectoryName(Configuration conf, final String serverName) {
325    StringBuilder dirName = new StringBuilder(HConstants.HREGION_OLDLOGDIR_NAME);
326    if (conf.getBoolean(SEPARATE_OLDLOGDIR, DEFAULT_SEPARATE_OLDLOGDIR)) {
327      dirName.append(Path.SEPARATOR);
328      dirName.append(serverName);
329    }
330    return dirName.toString();
331  }
332
333  /**
334   * List all the old wal files for a dead region server.
335   * <p/>
336   * Initially added for supporting replication, where we need to get the wal files to replicate for
337   * a dead region server.
338   */
339  public static List<Path> getArchivedWALFiles(Configuration conf, ServerName serverName,
340    String logPrefix) throws IOException {
341    Path walRootDir = CommonFSUtils.getWALRootDir(conf);
342    FileSystem fs = walRootDir.getFileSystem(conf);
343    List<Path> archivedWalFiles = new ArrayList<>();
344    // list both the root old wal dir and the separate old wal dir, so we will not miss any files if
345    // the SEPARATE_OLDLOGDIR config is changed
346    Path oldWalDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
347    try {
348      for (FileStatus status : fs.listStatus(oldWalDir, p -> p.getName().startsWith(logPrefix))) {
349        if (status.isFile()) {
350          archivedWalFiles.add(status.getPath());
351        }
352      }
353    } catch (FileNotFoundException e) {
354      LOG.info("Old WAL dir {} not exists", oldWalDir);
355      return Collections.emptyList();
356    }
357    Path separatedOldWalDir = new Path(oldWalDir, serverName.toString());
358    try {
359      for (FileStatus status : fs.listStatus(separatedOldWalDir,
360        p -> p.getName().startsWith(logPrefix))) {
361        if (status.isFile()) {
362          archivedWalFiles.add(status.getPath());
363        }
364      }
365    } catch (FileNotFoundException e) {
366      LOG.info("Seprated old WAL dir {} not exists", separatedOldWalDir);
367    }
368    return archivedWalFiles;
369  }
370
371  /**
372   * List all the wal files for a logPrefix.
373   */
374  public static List<Path> getWALFiles(Configuration c, ServerName serverName) throws IOException {
375    Path walRoot = new Path(CommonFSUtils.getWALRootDir(c), HConstants.HREGION_LOGDIR_NAME);
376    FileSystem fs = walRoot.getFileSystem(c);
377    List<Path> walFiles = new ArrayList<>();
378    Path walDir = new Path(walRoot, serverName.toString());
379    try {
380      for (FileStatus status : fs.listStatus(walDir)) {
381        if (status.isFile()) {
382          walFiles.add(status.getPath());
383        }
384      }
385    } catch (FileNotFoundException e) {
386      LOG.info("WAL dir {} not exists", walDir);
387    }
388    return walFiles;
389  }
390
391  /**
392   * Pulls a ServerName out of a Path generated according to our layout rules. In the below layouts,
393   * this method ignores the format of the logfile component. Current format: [base directory for
394   * hbase]/hbase/.logs/ServerName/logfile or [base directory for
395   * hbase]/hbase/.logs/ServerName-splitting/logfile Expected to work for individual log files and
396   * server-specific directories.
397   * @return null if it's not a log file. Returns the ServerName of the region server that created
398   *         this log file otherwise.
399   */
400  public static ServerName getServerNameFromWALDirectoryName(Configuration conf, String path)
401    throws IOException {
402    if (path == null || path.length() <= HConstants.HREGION_LOGDIR_NAME.length()) {
403      return null;
404    }
405
406    if (conf == null) {
407      throw new IllegalArgumentException("parameter conf must be set");
408    }
409
410    final String rootDir = conf.get(HConstants.HBASE_DIR);
411    if (rootDir == null || rootDir.isEmpty()) {
412      throw new IllegalArgumentException(HConstants.HBASE_DIR + " key not found in conf.");
413    }
414
415    final StringBuilder startPathSB = new StringBuilder(rootDir);
416    if (!rootDir.endsWith("/")) {
417      startPathSB.append('/');
418    }
419    startPathSB.append(HConstants.HREGION_LOGDIR_NAME);
420    if (!HConstants.HREGION_LOGDIR_NAME.endsWith("/")) {
421      startPathSB.append('/');
422    }
423    final String startPath = startPathSB.toString();
424
425    String fullPath;
426    try {
427      fullPath = FileSystem.get(conf).makeQualified(new Path(path)).toString();
428    } catch (IllegalArgumentException e) {
429      LOG.info("Call to makeQualified failed on " + path + " " + e.getMessage());
430      return null;
431    }
432
433    if (!fullPath.startsWith(startPath)) {
434      return null;
435    }
436
437    final String serverNameAndFile = fullPath.substring(startPath.length());
438
439    if (serverNameAndFile.indexOf('/') < "a,0,0".length()) {
440      // Either it's a file (not a directory) or it's not a ServerName format
441      return null;
442    }
443
444    Path p = new Path(path);
445    return getServerNameFromWALDirectoryName(p);
446  }
447
448  /**
449   * This function returns region server name from a log file name which is in one of the following
450   * formats:
451   * <ul>
452   * <li>hdfs://&lt;name node&gt;/hbase/.logs/&lt;server name&gt;-splitting/...</li>
453   * <li>hdfs://&lt;name node&gt;/hbase/.logs/&lt;server name&gt;/...</li>
454   * </ul>
455   * @return null if the passed in logFile isn't a valid WAL file path
456   */
457  public static ServerName getServerNameFromWALDirectoryName(Path logFile) {
458    String logDirName = logFile.getParent().getName();
459    // We were passed the directory and not a file in it.
460    if (logDirName.equals(HConstants.HREGION_LOGDIR_NAME)) {
461      logDirName = logFile.getName();
462    }
463    ServerName serverName = null;
464    if (logDirName.endsWith(SPLITTING_EXT)) {
465      logDirName = logDirName.substring(0, logDirName.length() - SPLITTING_EXT.length());
466    }
467    try {
468      serverName = ServerName.parseServerName(logDirName);
469    } catch (IllegalArgumentException | IllegalStateException ex) {
470      serverName = null;
471      LOG.warn("Cannot parse a server name from path={}", logFile, ex);
472    }
473    if (serverName != null && serverName.getStartCode() < 0) {
474      LOG.warn("Invalid log file path={}, start code {} is less than 0", logFile,
475        serverName.getStartCode());
476      serverName = null;
477    }
478    return serverName;
479  }
480
481  public static boolean isMetaFile(Path p) {
482    return isMetaFile(p.getName());
483  }
484
485  /** Returns True if String ends in {@link #META_WAL_PROVIDER_ID} */
486  public static boolean isMetaFile(String p) {
487    return p != null && p.endsWith(META_WAL_PROVIDER_ID);
488  }
489
490  /**
491   * Comparator used to compare WAL files together based on their start time. Just compares start
492   * times and nothing else.
493   */
494  public static class WALStartTimeComparator implements Comparator<Path> {
495    @Override
496    public int compare(Path o1, Path o2) {
497      return Long.compare(getTS(o1), getTS(o2));
498    }
499
500    /**
501     * Split a path to get the start time For example: 10.20.20.171%3A60020.1277499063250 Could also
502     * be a meta WAL which adds a '.meta' suffix or a synchronous replication WAL which adds a
503     * '.syncrep' suffix. Check.
504     * @param p path to split
505     * @return start time
506     */
507    public static long getTS(Path p) {
508      return getTimestamp(p.getName());
509    }
510  }
511
512  public static boolean isArchivedLogFile(Path p) {
513    String oldLog = Path.SEPARATOR + HConstants.HREGION_OLDLOGDIR_NAME + Path.SEPARATOR;
514    return p.toString().contains(oldLog);
515  }
516
517  /**
518   * Find the archived WAL file path if it is not able to locate in WALs dir.
519   * @param path - active WAL file path
520   * @param conf - configuration
521   * @return archived path if exists, null - otherwise
522   * @throws IOException exception
523   */
524  public static Path findArchivedLog(Path path, Configuration conf) throws IOException {
525    // If the path contains oldWALs keyword then exit early.
526    if (path.toString().contains(HConstants.HREGION_OLDLOGDIR_NAME)) {
527      return null;
528    }
529    Path walRootDir = CommonFSUtils.getWALRootDir(conf);
530    FileSystem fs = path.getFileSystem(conf);
531    // Try finding the log in old dir
532    Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
533    Path archivedLogLocation = new Path(oldLogDir, path.getName());
534    if (fs.exists(archivedLogLocation)) {
535      LOG.info("Log " + path + " was moved to " + archivedLogLocation);
536      return archivedLogLocation;
537    }
538
539    ServerName serverName = getServerNameFromWALDirectoryName(path);
540    if (serverName == null) {
541      LOG.warn("Can not extract server name from path {}, "
542        + "give up searching the separated old log dir", path);
543      return null;
544    }
545    // Try finding the log in separate old log dir
546    oldLogDir = new Path(walRootDir, new StringBuilder(HConstants.HREGION_OLDLOGDIR_NAME)
547      .append(Path.SEPARATOR).append(serverName.getServerName()).toString());
548    archivedLogLocation = new Path(oldLogDir, path.getName());
549    if (fs.exists(archivedLogLocation)) {
550      LOG.info("Log " + path + " was moved to " + archivedLogLocation);
551      return archivedLogLocation;
552    }
553    LOG.error("Couldn't locate log: " + path);
554    return null;
555  }
556
557  // For HBASE-15019
558  public static void recoverLease(Configuration conf, Path path) {
559    try {
560      final FileSystem dfs = CommonFSUtils.getCurrentFileSystem(conf);
561      RecoverLeaseFSUtils.recoverFileLease(dfs, path, conf, new CancelableProgressable() {
562        @Override
563        public boolean progress() {
564          LOG.debug("Still trying to recover WAL lease: " + path);
565          return true;
566        }
567      });
568    } catch (IOException e) {
569      LOG.warn("unable to recover lease for WAL: " + path, e);
570    }
571  }
572
573  @Override
574  public void addWALActionsListener(WALActionsListener listener) {
575    listeners.add(listener);
576  }
577
578  private static String getWALNameGroupFromWALName(String name, int group) {
579    Matcher matcher = WAL_FILE_NAME_PATTERN.matcher(name);
580    if (matcher.matches()) {
581      return matcher.group(group);
582    } else {
583      throw new IllegalArgumentException(name + " is not a valid wal file name");
584    }
585  }
586
587  /**
588   * Get prefix of the log from its name, assuming WAL name in format of
589   * log_prefix.filenumber.log_suffix
590   * @param name Name of the WAL to parse
591   * @return prefix of the log
592   * @throws IllegalArgumentException if the name passed in is not a valid wal file name
593   * @see AbstractFSWAL#getCurrentFileName()
594   */
595  public static String getWALPrefixFromWALName(String name) {
596    return getWALNameGroupFromWALName(name, 1);
597  }
598
599  private static final Pattern SERVER_NAME_PATTERN = Pattern.compile("^[^"
600    + ServerName.SERVERNAME_SEPARATOR + "]+" + ServerName.SERVERNAME_SEPARATOR
601    + Addressing.VALID_PORT_REGEX + ServerName.SERVERNAME_SEPARATOR + Addressing.VALID_PORT_REGEX);
602
603  /**
604   * Parse the server name from wal prefix. A wal's name is always started with a server name in non
605   * test code.
606   * @throws IllegalArgumentException if the name passed in is not started with a server name
607   * @return the server name
608   */
609  public static ServerName parseServerNameFromWALName(String name) {
610    String decoded;
611    try {
612      decoded = URLDecoder.decode(name, StandardCharsets.UTF_8.name());
613    } catch (UnsupportedEncodingException e) {
614      throw new AssertionError("should never happen", e);
615    }
616    Matcher matcher = SERVER_NAME_PATTERN.matcher(decoded);
617    if (matcher.find()) {
618      return ServerName.valueOf(matcher.group());
619    } else {
620      throw new IllegalArgumentException(name + " is not started with a server name");
621    }
622  }
623}