[GOBBLIN-264] Add a SharedResourceFactory for creating shared DataPub…

Closes apache#2116 from htran1/shareable_publishers

Author: htran1

gobblin-api/src/main/java/org/apache/gobblin/capability/Capability.java

@@ -0,0 +1,40 @@
+/*
+ * 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.gobblin.capability;
+
+import org.apache.gobblin.annotation.Alpha;
+
+import lombok.Data;
+
+/**
+ * Represents a set of functionality a job-creator can ask for. Examples could include
+ * encryption, compression, partitioning...
+ *
+ * Each Capability has a name and then a set of associated configuration properties. An example is
+ * the encryption algorithm to use.
+ */
+@Alpha
+@Data
+public class Capability {
+  /**
+   * Threadsafe capability.
+   */
+  public static final Capability THREADSAFE = new Capability("THREADSAFE", false);
+
+  private final String name;
+  private final boolean critical;
+}

gobblin-api/src/main/java/org/apache/gobblin/capability/CapabilityAware.java

@@ -0,0 +1,38 @@
+/*
+ * 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.gobblin.capability;
+
+import java.util.Map;
+
+import org.apache.gobblin.annotation.Alpha;
+
+/**
+ * Describes an object that is aware of the capabilities it supports.
+ */
+@Alpha
+public interface CapabilityAware {
+  /**
+   * Checks if this object supports the given Capability with the given properties.
+   *
+   * Implementers of this should always check if their super-class may happen to support a capability
+   * before returning false!
+   * @param c Capability being queried
+   * @param properties Properties specific to the capability. Properties are capability specific.
+   * @return True if this object supports the given capability + property settings, false if not
+   */
+  boolean supportsCapability(Capability c, Map<String, Object> properties);
+}

gobblin-api/src/main/java/org/apache/gobblin/publisher/DataPublisher.java

@@ -21,7 +21,10 @@
 import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.util.Collection;
+import java.util.Map;
 
+import org.apache.gobblin.capability.Capability;
+import org.apache.gobblin.capability.CapabilityAware;
 import org.apache.gobblin.configuration.ConfigurationKeys;
 import org.apache.gobblin.configuration.State;
 import org.apache.gobblin.configuration.WorkUnitState;
@@ -30,7 +33,11 @@
 /**
  * Defines how to publish data and its corresponding metadata. Can be used for either task level or job level publishing.
  */
-public abstract class DataPublisher implements Closeable {
+public abstract class DataPublisher implements Closeable, CapabilityAware {
+  /**
+   * Reusable capability.
+   */
+  public static final Capability REUSABLE = new Capability("REUSABLE", false);
 
   protected final State state;
 
@@ -125,4 +132,9 @@ public boolean canBeSkipped() {
   protected boolean shouldPublishMetadataFirst() {
     return true;
   }
+
+  @Override
+  public boolean supportsCapability(Capability c, Map<String, Object> properties) {
+    return false;
+  }
 }

gobblin-core/src/main/java/org/apache/gobblin/publisher/DataPublisherFactory.java

@@ -0,0 +1,92 @@
+/*
+ * 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.gobblin.publisher;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import org.apache.gobblin.broker.ImmediatelyInvalidResourceEntry;
+import org.apache.gobblin.broker.ResourceInstance;
+import org.apache.gobblin.broker.iface.ConfigView;
+import org.apache.gobblin.broker.iface.NotConfiguredException;
+import org.apache.gobblin.broker.iface.ScopeType;
+import org.apache.gobblin.broker.iface.ScopedConfigView;
+import org.apache.gobblin.broker.iface.SharedResourceFactory;
+import org.apache.gobblin.broker.iface.SharedResourceFactoryResponse;
+import org.apache.gobblin.broker.iface.SharedResourcesBroker;
+import org.apache.gobblin.capability.Capability;
+import org.apache.gobblin.configuration.State;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * A {@link SharedResourceFactory} for creating {@link DataPublisher}s.
+ *
+ * The factory creates a {@link DataPublisher} with the publisher class name and state.
+ */
+@Slf4j
+public class DataPublisherFactory<S extends ScopeType<S>>
+    implements SharedResourceFactory<DataPublisher, DataPublisherKey, S> {
+
+  public static final String FACTORY_NAME = "dataPublisher";
+
+  public static <S extends ScopeType<S>> DataPublisher get(String publisherClassName, State state,
+      SharedResourcesBroker<S> broker) throws IOException {
+    try {
+      return broker.getSharedResource(new DataPublisherFactory<S>(), new DataPublisherKey(publisherClassName, state));
+    } catch (NotConfiguredException nce) {
+      throw new IOException(nce);
+    }
+  }
+
+  @Override
+  public String getName() {
+    return FACTORY_NAME;
+  }
+
+  @Override
+  public SharedResourceFactoryResponse<DataPublisher> createResource(SharedResourcesBroker<S> broker,
+      ScopedConfigView<S, DataPublisherKey> config) throws NotConfiguredException {
+    try {
+      DataPublisherKey key = config.getKey();
+      String publisherClassName = key.getPublisherClassName();
+      State state = key.getState();
+      Class<? extends DataPublisher> dataPublisherClass =  (Class<? extends DataPublisher>) Class
+          .forName(publisherClassName);
+
+      DataPublisher publisher = DataPublisher.getInstance(dataPublisherClass, state);
+
+      // If the publisher is threadsafe then it is shareable, so return it as a resource instance that may be cached
+      // by the broker.
+      // Otherwise, it is not shareable, so return it as an immediately invalidated resource that will only be returned
+      // once from the broker.
+      if (publisher.supportsCapability(Capability.THREADSAFE, Collections.EMPTY_MAP)) {
+        return new ResourceInstance<>(publisher);
+      } else {
+        return new ImmediatelyInvalidResourceEntry<>(publisher);
+      }
+    } catch (ReflectiveOperationException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public S getAutoScope(SharedResourcesBroker<S> broker, ConfigView<S, DataPublisherKey> config) {
+    return broker.selfScope().getType().rootScope();
+  }
+}

gobblin-core/src/main/java/org/apache/gobblin/publisher/DataPublisherKey.java

@@ -0,0 +1,64 @@
+/*
+ * 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.gobblin.publisher;
+
+import org.apache.gobblin.broker.iface.SharedResourceKey;
+import org.apache.gobblin.configuration.State;
+
+import lombok.Getter;
+
+
+/**
+ * {@link SharedResourceKey} for requesting {@link DataPublisher}s from a
+ * {@link org.apache.gobblin.broker.iface.SharedResourceFactory
+ */
+@Getter
+public class DataPublisherKey implements SharedResourceKey {
+  private final String publisherClassName;
+  private final State state;
+
+  public DataPublisherKey(String publisherClassName, State state) {
+    this.publisherClassName = publisherClassName;
+    this.state = state;
+  }
+
+  @Override
+  public String toConfigurationKey() {
+    return this.publisherClassName;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    DataPublisherKey that = (DataPublisherKey) o;
+
+    return publisherClassName == null ?
+        that.publisherClassName == null : publisherClassName.equals(that.publisherClassName);
+  }
+
+  @Override
+  public int hashCode() {
+    return publisherClassName != null ? publisherClassName.hashCode() : 0;
+  }
+}