mattcasters commented on code in PR #8819:
URL: https://github.com/apache/hop/pull/8819#discussion_r4233900653
##########
plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java:
##########
@@ -40,7 +68,268 @@ public LakeTableOutput(
@Override
public boolean processRow() throws HopException {
- setOutputDone();
+ try (PluginClassLoader ignored = PluginClassLoader.activate()) {
+ return writeRow();
+ }
+ }
+
+ private boolean writeRow() throws HopException {
+ Object[] row = getRow();
+ if (first) {
+ first = false;
+ IRowMeta rowMeta =
+ row == null
+ ? getPipelineMeta().getPrevTransformFields(this,
getTransformMeta())
+ : getInputRowMeta();
+ data.coordinator = coordinator(rowMeta).orElse(null);
+ if (data.coordinator != null) {
+ data.writer =
+ new IcebergRowWriter(
+ data.coordinator.table(),
+ rowMeta,
+ getCopy(),
+ System.currentTimeMillis(),
+ IcebergRowWriter.DEFAULT_TARGET_FILE_SIZE);
+ }
+ }
+
+ if (row == null) {
+ if (data.writer != null) {
+ data.coordinator.addFiles(data.writer.complete());
+ data.writer = null;
+ }
+ setOutputDone();
+ return false;
+ }
+
+ if (data.writer != null) {
+ data.writer.write(row);
+ }
+ putRow(getInputRowMeta(), row);
+ return true;
+ }
+
+ /**
+ * The commit coordinator shared by all copies of this transform in this
run, created by the first
+ * copy that gets here. Empty when the save mode says to leave an existing
table alone.
+ */
+ @SuppressWarnings("unchecked")
+ private Optional<CommitCoordinator> coordinator(IRowMeta rowMeta) throws
HopException {
+ Map<String, Object> shared = getPipeline().getExtensionDataMap();
+ String key = COORDINATOR_KEY + getTransformName();
+ synchronized (shared) {
+ if (!shared.containsKey(key)) {
+ Optional<CommitCoordinator> coordinator =
Optional.ofNullable(newCoordinator(rowMeta));
+ shared.put(key, coordinator);
+ coordinator.ifPresent(this::commitWhenPipelineFinishes);
+ }
+ return (Optional<CommitCoordinator>) shared.get(key);
+ }
+ }
+
+ /** Coalesce and extra options configure a Spark write; this engine doesn't
use them. */
+ private void warnAboutSparkOnlyOptions() {
+ for (String[] option :
+ new String[][] {
+ {"Coalesce partitions", meta.getCoalescePartitions()},
+ {"Extra options", meta.getExtraOptions()}
+ }) {
+ if (StringUtils.isNotBlank(resolve(option[1]))) {
+ logError(
+ "Warning: '"
+ + option[0]
+ + "' only applies to the Spark engine and is ignored when
writing on this engine");
+ }
+ }
+ }
+
+ private CommitCoordinator newCoordinator(IRowMeta rowMeta) throws
HopException {
+ warnAboutSparkOnlyOptions();
+ String format = resolve(meta.getFormat());
+ if (!LakeFormats.FORMAT_ICEBERG.equalsIgnoreCase(format)) {
+ throw new HopException(
+ "Table format '"
+ + format
+ + "' isn't supported on this engine yet. The local engine writes
Apache Iceberg"
+ + " tables; Delta Lake tables can be written with the Spark
engine.");
+ }
+
+ IcebergTableTarget target = target();
+ String saveMode =
+ StringUtils.defaultIfBlank(resolve(meta.getSaveMode()),
LakeTableOutputMeta.MODE_ERROR);
+ Map<String, String> summary = new HashMap<>();
+ summary.put(CommitCoordinator.SUMMARY_RUN_ID,
getPipeline().getLogChannelId());
+ summary.put("hop.pipeline", Const.NVL(getPipelineMeta().getName(), ""));
+
+ if (target.exists()) {
+ if (LakeTableOutputMeta.MODE_ERROR.equalsIgnoreCase(saveMode)) {
+ throw new HopException(
+ "Table "
+ + target.name()
+ + " already exists. Choose the Append or Overwrite save mode
to write to it.");
+ }
+ if (LakeTableOutputMeta.MODE_IGNORE.equalsIgnoreCase(saveMode)) {
+ logBasic("Table " + target.name() + " already exists, rows are not
written (Ignore)");
+ return null;
+ }
+ WriteMode mode;
+ if (LakeTableOutputMeta.MODE_OVERWRITE.equalsIgnoreCase(saveMode)) {
+ mode = WriteMode.OVERWRITE_TABLE;
+ } else if (LakeTableOutputMeta.MODE_APPEND.equalsIgnoreCase(saveMode)) {
+ mode = WriteMode.APPEND;
+ } else {
+ throw unsupportedSaveMode(saveMode);
+ }
+ return new CommitCoordinator(target.load(), mode, summary);
+ }
+
+ if (!isSupportedSaveMode(saveMode)) {
+ throw unsupportedSaveMode(saveMode);
+ }
+ Schema schema = IcebergTypeMapper.toIcebergSchema(rowMeta);
+ PartitionSpec spec = partitionSpec(schema,
resolve(meta.getPartitionByColumns()));
+ logBasic(
+ "Creating table "
+ + target.name()
+ + (spec.isUnpartitioned() ? "" : ", partitioned by " +
meta.getPartitionByColumns()));
+ return new CommitCoordinator(target.create(schema, spec, Map.of()),
summary);
+ }
+
+ private static boolean isSupportedSaveMode(String saveMode) {
+ for (String mode :
+ new String[] {
+ LakeTableOutputMeta.MODE_APPEND,
+ LakeTableOutputMeta.MODE_OVERWRITE,
+ LakeTableOutputMeta.MODE_ERROR,
+ LakeTableOutputMeta.MODE_IGNORE
+ }) {
+ if (mode.equalsIgnoreCase(saveMode)) {
+ return true;
+ }
+ }
return false;
}
+
+ private static HopException unsupportedSaveMode(String saveMode) {
+ return new HopException(
+ "Save mode '"
+ + saveMode
+ + "' isn't supported. Use Append, Overwrite, ErrorIfExists or
Ignore.");
+ }
+
+ private IcebergTableTarget target() throws HopException {
+ if
(LakeTableInputMeta.MODE_TABLE.equalsIgnoreCase(meta.getIdentifierMode())) {
+ String catalogName = resolve(meta.getCatalogMetadataName());
+ if (StringUtils.isBlank(catalogName)) {
+ throw new HopException("Table mode needs a catalog: please select
one");
+ }
+ LakeCatalog catalog =
metadataProvider.getSerializer(LakeCatalog.class).load(catalogName);
+ if (catalog == null) {
+ throw new HopException("Catalog '" + catalogName + "' couldn't be
found");
+ }
+ return IcebergTables.targetInCatalog(catalog, meta.getTableIdentifier(),
this);
+ }
+ String path = resolve(meta.getTablePath());
+ if (StringUtils.isBlank(path)) {
+ throw new HopException("Please specify the location of the table");
+ }
+ return IcebergTableTarget.atPath(path);
+ }
+
+ /** Identity partitions on the listed columns, used when the table is
created. */
+ static PartitionSpec partitionSpec(Schema schema, String columns) throws
HopException {
+ if (StringUtils.isBlank(columns)) {
+ return PartitionSpec.unpartitioned();
+ }
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(schema);
+ List<String> missing = new ArrayList<>();
+ for (String column : columns.split(",")) {
+ String name = column.trim();
+ if (name.isEmpty()) {
+ continue;
+ }
+ if (schema.findField(name) == null) {
+ missing.add(name);
+ } else {
+ builder.identity(name);
+ }
+ }
+ if (!missing.isEmpty()) {
+ throw new HopException("Partition column(s) not found in the input: " +
missing);
+ }
+ return builder.build();
+ }
+
+ private void commitWhenPipelineFinishes(CommitCoordinator coordinator) {
+ getPipeline()
+ .addExecutionFinishedListener(
Review Comment:
**[bug]** `addExecutionFinishedListener` appends, and
`Pipeline.startThreads` has already inserted its own listener at index 0. That
listener calls `setRunning(false)` and then `cleanupRowSets()` and
`HopVfs.freeUnusedResources` before this one runs. `Pipeline.getStatus()` is
therefore `Finished` with zero errors for that whole stretch.
`HopServerPipelineStatus.isFinished()` and `RemotePipelineEngine` key off that
description: the remote engine stores `getResult()` as soon as it sees
`Finished`, and its `waitUntilFinished()` returns. `setRunning(true)` inside
`finishRun` is too late to cover the gap, so a parent workflow on the remote
engine (which this change enables) can start the next action before the
snapshot exists, and a commit that fails afterwards is missing from the result
it already stored. The local `waitUntilFinished` path is fine:
`pipelineCompleted()` runs only after every listener returns, and `setErrors`
happens before that.
**Suggestion:** On the local `Pipeline`, insert this listener at index 0
after `startThreads` (the list from `getExecutionFinishedListeners()` is live)
so the commit runs while `getStatus()` is still `Running`, before the built-in
listener flips the status and frees VFS. Keep recording the transform error
before returning.
##########
plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/PathTableOperations.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.hop.lakehouse.iceberg;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.FileSystemException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.commons.vfs2.FileObject;
+import org.apache.commons.vfs2.provider.local.LocalFile;
+import org.apache.hop.core.logging.LogChannel;
+import org.apache.hop.core.vfs.HopVfs;
+import org.apache.hop.lakehouse.iceberg.io.HopVfsFileIO;
+import org.apache.iceberg.LocationProviders;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+
+/**
+ * Table operations for an Iceberg table that lives in a folder without a
catalog, in the layout of
+ * Hadoop catalogs and Spark path tables: {@code metadata/v<N>.metadata.json}
plus a {@code
+ * version-hint.text} that names the current version.
+ *
+ * <p>A commit writes the new metadata to a temporary file and then publishes
it under the next
+ * version's name with an operation that fails if that name is already taken.
On a local file system
+ * that is a hard link, which the operating system creates atomically, so two
writers can't both
+ * commit the same version, whether they run in one Hop instance or in several
processes. Commits in
+ * one JVM are also serialized per table location. Other file systems publish
with a check followed
+ * by a write, which isn't atomic across processes, so a path table there
(including HDFS, object
+ * stores and local file systems without hard links) should have one writer at
a time. Tables
+ * managed by a catalog should be written through the catalog instead. Nothing
in a commit ever
+ * deletes or replaces an existing version file.
+ *
+ * <p>The version hint is best effort. Once the metadata file is in place the
commit has happened,
+ * so a failure to update the hint is only logged, and readers walk forward
from the hint to the
+ * newest version (see {@link IcebergTables#findCurrentMetadataFile(String)}).
+ */
+public class PathTableOperations implements TableOperations {
+
+ /** Serializes commits to the same table location within this JVM. */
+ private static final Map<String, ReentrantLock> LOCKS = new
ConcurrentHashMap<>();
+
+ private final String location;
+ private final FileIO io = new HopVfsFileIO();
+ private TableMetadata current;
+ private int version = -1;
+
+ public PathTableOperations(String location) {
+ this.location =
+ location.endsWith("/") ? location.substring(0, location.length() - 1)
: location;
+ }
+
+ @Override
+ public TableMetadata current() {
+ if (current == null && version < 0) {
+ refresh();
+ }
+ return current;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ String metadataFile;
+ try {
+ metadataFile = IcebergTables.findCurrentMetadataFile(location);
+ } catch (Exception e) {
+ throw new RuntimeIOException(
+ new java.io.IOException("Unable to read Iceberg table at " +
location, e));
+ }
+ if (metadataFile == null) {
+ // No table yet: it is created by the first commit.
+ version = 0;
+ current = null;
+ } else {
+ version =
+ (int)
+ IcebergTables.metadataVersion(
+ metadataFile.substring(metadataFile.lastIndexOf('/') + 1));
+ current = TableMetadataParser.read(io, metadataFile);
+ }
+ return current;
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ ReentrantLock lock = LOCKS.computeIfAbsent(location, ignored -> new
ReentrantLock());
+ lock.lock();
+ try {
+ commitLocked(base, metadata);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void commitLocked(TableMetadata base, TableMetadata metadata) {
+ TableMetadata latest = refresh();
+ if (base != latest
+ && (base == null
+ || latest == null
+ ||
!base.metadataFileLocation().equals(latest.metadataFileLocation()))) {
+ throw new CommitFailedException(
+ "Table %s was changed by another writer since it was read",
location);
+ }
+
+ int next = version + 1;
+ String target = metadataFileLocation("v" + next + ".metadata.json");
+ String temp = metadataFileLocation(UUID.randomUUID() +
".metadata.json.tmp");
+ TableMetadataParser.write(metadata, io.newOutputFile(temp));
+ try {
+ publish(temp, target, next);
+ } finally {
+ deleteQuietly(temp);
+ }
+
+ // The commit has happened. Nothing after this point may fail it.
+ try {
+ writeVersionHint(next);
+ } catch (Exception e) {
+ LogChannel.GENERAL.logBasic(
+ "Committed version "
+ + next
+ + " of Iceberg table "
+ + location
+ + ", but couldn't update version-hint.text; readers find the new
version anyway: "
+ + e.getMessage());
+ }
+ // Read the new metadata lazily, on the next call to current() or
refresh().
+ current = null;
+ version = -1;
+ }
+
+ /**
+ * Publishes {@code temp} as {@code target}, failing with a {@link
CommitFailedException} if
+ * another writer committed that version first. Nothing here ever deletes or
replaces an existing
+ * version file.
+ *
+ * <p>On a local file system with hard links, the link is atomic, so this
also holds between
+ * processes. Everywhere else (HDFS, object stores, local file systems
without hard links) the
+ * check and the write are separate steps, which is why path tables there
support one writer at a
+ * time. A failure while writing the version file there is reported as a
{@link
+ * CommitStateUnknownException}, since the file may already be visible.
+ */
+ void publish(String temp, String target, int next) {
+ FileObject targetFile;
+ try {
+ FileObject tempFile = HopVfs.getFileObject(temp);
+ targetFile = HopVfs.getFileObject(target);
+ if (tempFile instanceof LocalFile && targetFile instanceof LocalFile) {
+ try {
+ // link(2) fails with EEXIST if the name is taken, atomically.
+ Files.createLink(Paths.get(targetFile.getURI()),
Paths.get(tempFile.getURI()));
Review Comment:
**[bug]** `Files.createLink` is given `Paths.get(fileObject.getURI())`.
`FileObject.getURI()` is a URI (`file:///...`), and `Paths.get(String)` treats
that as a relative path (`file:/...`), not a URI. `createLink` throws
`NoSuchFileException`, which is a `FileSystemException` and is swallowed here,
so the atomic hard-link publish never runs on a local disk. Every commit falls
through to the `exists` check plus `HopVfs.getOutputStream(target, false)`.
That stream deletes the destination when it already exists, so two processes
that both pass the check can replace each other's `v<N>.metadata.json`. The
in-JVM lock does not cover that. A failed `transferTo` also leaves a partial
version file, and `findCurrentMetadataFile` will then treat it as the current
metadata.
**Suggestion:** Build the link paths with `Path.of(URI.create(uri))` or
`HopVfs.getFilename` for a `LocalFile`, and fall through only on
`UnsupportedOperationException` or a real "no hard links" error, not
`NoSuchFileException`. On the copy path, fail the commit instead of opening the
destination with `getOutputStream` (it deletes an existing file), and delete a
partial `v<N>.metadata.json` when the write fails before completion.
##########
plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/CommitCoordinator.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.hop.lakehouse.iceberg;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.ReplacePartitions;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotUpdate;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * Collects the data files written by every copy of an output transform and
turns them into a single
+ * snapshot, or into nothing at all.
+ *
+ * <p>Copies call {@link #addFiles(List)} as they finish. Once the whole
pipeline is done, exactly
+ * one of {@link #commit()} or {@link #abort()} is called: commit when the run
ended without errors,
+ * abort otherwise. This keeps a run atomic. Readers either see all of its
rows or none of them,
+ * even when a transform further down the pipeline fails after this one has
finished writing.
+ */
+public class CommitCoordinator {
+
+ public enum WriteMode {
+ APPEND,
+ OVERWRITE_TABLE,
+ OVERWRITE_PARTITIONS
+ }
+
+ public static final String SUMMARY_RUN_ID = "hop.run-id";
+
+ private final Table table;
+ private final Transaction createTransaction;
+ private final WriteMode writeMode;
+ private final Map<String, String> summary;
+ private final Long startSnapshotId;
+ private boolean published;
+ private boolean outcomeUnknown;
+ private final List<DataFile> files = new ArrayList<>();
+ private boolean done;
+
+ /**
+ * Commits to an existing table.
+ *
+ * @param summary extra properties for the snapshot summary, such as the
pipeline name and run id
+ */
+ public CommitCoordinator(Table table, WriteMode writeMode, Map<String,
String> summary) {
+ this(table, null, writeMode, summary);
+ }
+
+ /**
+ * Commits to a table that is created by this run. The table only comes into
existence together
+ * with the run's snapshot, so a failed run leaves no empty table behind.
+ */
+ public CommitCoordinator(Transaction createTransaction, Map<String, String>
summary) {
+ this(createTransaction.table(), createTransaction, WriteMode.APPEND,
summary);
+ }
+
+ private CommitCoordinator(
+ Table table,
+ Transaction createTransaction,
+ WriteMode writeMode,
+ Map<String, String> summary) {
+ this.table = table;
+ this.createTransaction = createTransaction;
+ this.writeMode = writeMode;
+ this.summary = Map.copyOf(summary);
+ Snapshot current = createTransaction == null ? table.currentSnapshot() :
null;
+ this.startSnapshotId = current == null ? null : current.snapshotId();
+ }
+
+ /** The table to write data files for. */
+ public Table table() {
+ return table;
+ }
+
+ public synchronized void addFiles(List<DataFile> dataFiles) {
+ if (done) {
+ throw new IllegalStateException("The commit for this run has already
been decided");
+ }
+ files.addAll(dataFiles);
+ }
+
+ /**
+ * Commits all collected files as one snapshot.
+ *
+ * @return the id of the new snapshot, or null if the commit happened but
the table couldn't be
+ * read back afterwards
+ * @throws CommitStateUnknownException if the catalog couldn't say whether
the commit happened;
+ * the data files must then be kept, see {@link #isPublished()}
+ */
+ public synchronized Long commit() {
+ done = true;
+ SnapshotUpdate<?> update =
+ switch (writeMode) {
+ case APPEND -> {
+ AppendFiles append = table.newAppend();
+ files.forEach(append::appendFile);
+ yield append;
+ }
+ case OVERWRITE_TABLE -> {
+ OverwriteFiles overwrite =
+
table.newOverwrite().overwriteByRowFilter(Expressions.alwaysTrue());
+ files.forEach(overwrite::addFile);
+ if (startSnapshotId != null) {
Review Comment:
**[bug]** Overwrite conflict checks are skipped when the table has no
snapshot yet (`startSnapshotId == null`). `SnapshotProducer.commit` retries a
`CommitFailedException` by refreshing and reapplying the overwrite. A
concurrent append that creates the table's first snapshot is then removed by
`overwriteByRowFilter(Expressions.alwaysTrue())` instead of failing the commit.
Tables that already have a snapshot are protected; an existing empty table is
not.
**Suggestion:** Always call `validateNoConflictingData()` for an overwrite.
With a null starting snapshot, Iceberg treats every current file as added since
the beginning (`ancestorsBetween` with a null oldest id), so a snapshot that
appeared during the run fails validation, while a table that is still empty
does not. `ValidationException` is not retried, so the run aborts and the other
writer's files stay.
##########
plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTables.java:
##########
@@ -221,18 +259,24 @@ static String currentMetadataFile(String location) throws
HopException {
newest = name;
}
}
- if (newest == null) {
- throw new HopException(
- "No Iceberg table found at '" + root + "': the metadata folder has
no metadata files");
- }
- return metadataFolder + "/" + newest;
- } catch (HopException e) {
- throw e;
+ return newest == null ? null : metadataFolder + "/" + newest;
} catch (Exception e) {
throw new HopException("Unable to find the Iceberg metadata of table '"
+ root + "'", e);
}
}
+ /** The metadata file of {@code version} in a path-table layout, or null if
it doesn't exist. */
+ private static String versionFile(String metadataFolder, long version)
throws Exception {
+ for (String name :
+ new String[] {"v" + version + ".metadata.json", "v" + version +
".gz.metadata.json"}) {
Review Comment:
**[suggestion]** The forward walk only looks for `v<N>.metadata.json` and
`v<N>.gz.metadata.json`. Iceberg's `HadoopTableOperations.getMetadataFile` also
accepts the legacy gzip name `v<N>.metadata.json.gz`
(`TableMetadataParser.getOldFileExtension`). A stale hint in front of one of
those files stops the walk early, and the next commit can publish a new
`v<N>.metadata.json` that hides the gzip version. `metadataVersion` does not
match that name either, so the folder-listing fallback will not correct it.
**Suggestion:** Also probe `v<N>.metadata.json.gz` in `versionFile`, and let
`metadataVersion` recognize that name.
--
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]
