modules/kafka-ext/src/main/java/org/apache/ignite/stream/kafka/connect/IgniteSourceTask.java - ignite-extensions - Git at Google

 /*
  * 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.ignite.stream.kafka.connect;

 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.Ignition;
 import org.apache.ignite.cache.affinity.Affinity;
 import org.apache.ignite.events.CacheEvent;
 import org.apache.ignite.events.EventType;
 import org.apache.ignite.lang.IgniteBiPredicate;
 import org.apache.ignite.lang.IgnitePredicate;
 import org.apache.ignite.resources.IgniteInstanceResource;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * Task to consume remote cluster cache events from the grid and inject them into Kafka.
  * <p>
  * Note that a task will create a bounded queue in the grid for more reliable data transfer.
  * Queue size can be changed by {@link IgniteSourceConstants#INTL_BUF_SIZE}.
  */
 public class IgniteSourceTask extends SourceTask {
     /** Logger. */
     private static final Logger log = LoggerFactory.getLogger(IgniteSourceTask.class);

     /** Tasks static monitor. */
     private static final Object lock = new Object();

     /** Event buffer size. */
     private static int evtBufSize = 100000;

     /** Event buffer. */
     private static BlockingQueue<CacheEvent> evtBuf = new LinkedBlockingQueue<>(evtBufSize);

     /** Max number of events taken from the buffer at once. */
     private static int evtBatchSize = 100;

     /** Flag for stopped state. */
     private static volatile boolean stopped = true;

     /** Ignite grid configuration file. */
     private static String igniteCfgFile;

     /** Cache name. */
     private static String cacheName;

     /** Remote Listener id. */
     private static UUID rmtLsnrId;

     /** Local listener. */
     private static TaskLocalListener locLsnr = new TaskLocalListener();

     /** User-defined filter. */
     private static IgnitePredicate<CacheEvent> filter;

     /** Topic. */
     private static String topics[];

     /** Offset. */
     private static final Map<String, Long> offset = Collections.singletonMap("offset", 0L);

     /** Partition. */
     private static final Map<String, String> srcPartition = Collections.singletonMap("cache", null);

     /** {@inheritDoc} */
     @Override public String version() {
         return new IgniteSinkConnector().version();
     }

     /**
      * Filtering is done remotely. Local listener buffers data for injection into Kafka.
      *
      * @param props Task properties.
      */
     @Override public void start(Map<String, String> props) {
         synchronized (lock) {
             // Each task has the same parameters -- avoid setting more than once.
             // Nothing to do if the task has been already started.
             if (!stopped)
                 return;

             cacheName = props.get(IgniteSourceConstants.CACHE_NAME);
             igniteCfgFile = props.get(IgniteSourceConstants.CACHE_CFG_PATH);
             topics = props.get(IgniteSourceConstants.TOPIC_NAMES).split("\\s*,\\s*");

             if (props.containsKey(IgniteSourceConstants.INTL_BUF_SIZE))
                 evtBufSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BUF_SIZE));

             if (props.containsKey(IgniteSourceConstants.INTL_BATCH_SIZE))
                 evtBatchSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BATCH_SIZE));

             if (props.containsKey(IgniteSourceConstants.CACHE_FILTER_CLASS)) {
                 String filterCls = props.get(IgniteSourceConstants.CACHE_FILTER_CLASS);
                 if (filterCls != null && !filterCls.isEmpty()) {
                     try {
                         Class<? extends IgnitePredicate<CacheEvent>> clazz =
                             (Class<? extends IgnitePredicate<CacheEvent>>)Class.forName(filterCls);

                         filter = clazz.newInstance();
                     }
                     catch (Exception e) {
                         log.error("Failed to instantiate the provided filter! " +
                             "User-enabled filtering is ignored!", e);
                     }
                 }
             }

             TaskRemoteFilter rmtLsnr = new TaskRemoteFilter(cacheName);

             try {
                 int[] evts = cacheEvents(props.get(IgniteSourceConstants.CACHE_EVENTS));

                 rmtLsnrId = IgniteGrid.getIgnite().events(IgniteGrid.getIgnite().cluster().forCacheNodes(cacheName))
                     .remoteListen(locLsnr, rmtLsnr, evts);
             }
             catch (Exception e) {
                 log.error("Failed to register event listener!", e);

                 throw new ConnectException(e);
             }
             finally {
                 stopped = false;
             }
         }
     }

     /** {@inheritDoc} */
     @Override public List<SourceRecord> poll() throws InterruptedException {
         ArrayList<SourceRecord> records = new ArrayList<>(evtBatchSize);
         ArrayList<CacheEvent> evts = new ArrayList<>(evtBatchSize);

         if (stopped)
             return records;

         try {
             if (evtBuf.drainTo(evts, evtBatchSize) > 0) {
                 for (CacheEvent evt : evts) {
                     // schema and keys are ignored.
                     for (String topic : topics)
                         records.add(new SourceRecord(srcPartition, offset, topic, null, evt));
                 }

                 return records;
             }
         }
         catch (IgniteException e) {
             log.error("Error when polling event queue!", e);
         }

         // for shutdown.
         return null;
     }

     /**
      * Converts comma-delimited cache events strings to Ignite internal representation.
      *
      * @param evtPropsStr Comma-delimited cache event names.
      * @return Ignite internal representation of cache events to be registered with the remote listener.
      * @throws Exception If error.
      */
     private int[] cacheEvents(String evtPropsStr) throws Exception {
         String[] evtStr = evtPropsStr.split("\\s*,\\s*");

         if (evtStr.length == 0)
             return EventType.EVTS_CACHE;

         int[] evts = new int[evtStr.length];

         try {
             for (int i = 0; i < evtStr.length; i++)
                 evts[i] = CacheEvt.valueOf(evtStr[i].toUpperCase()).getId();
         }
         catch (Exception e) {
             log.error("Failed to recognize the provided cache event!", e);

             throw new Exception(e);
         }
         return evts;
     }

     /**
      * Stops the grid client.
      */
     @Override public synchronized void stop() {
         if (stopped)
             return;

         stopped = true;

         stopRemoteListen();

         IgniteGrid.getIgnite().close();
     }

     /**
      * Stops the remote listener.
      */
     protected void stopRemoteListen() {
         if (rmtLsnrId != null)
             IgniteGrid.getIgnite().events(IgniteGrid.getIgnite().cluster().forCacheNodes(cacheName))
                 .stopRemoteListen(rmtLsnrId);

         rmtLsnrId = null;
     }

     /**
      * Used by unit test to avoid restart node and valid state of the <code>stopped</code> flag.
      *
      * @param stopped Stopped flag.
      */
     protected static void setStopped(boolean stopped) {
         IgniteSourceTask.stopped = stopped;
     }

     /**
      * Local listener buffering cache events to be further sent to Kafka.
      */
     private static class TaskLocalListener implements IgniteBiPredicate<UUID, CacheEvent> {
         /** {@inheritDoc} */
         @Override public boolean apply(UUID id, CacheEvent evt) {
             try {
                 if (!evtBuf.offer(evt, 10, TimeUnit.MILLISECONDS))
                     log.error("Failed to buffer event {}", evt.name());
             }
             catch (InterruptedException e) {
                 e.printStackTrace();
             }

             return true;
         }
     }

     /**
      * Remote filter.
      */
     private static class TaskRemoteFilter implements IgnitePredicate<CacheEvent> {
         /** */
         @IgniteInstanceResource
         Ignite ignite;

         /** Cache name. */
         private final String cacheName;

         /**
          * @param cacheName Cache name.
          */
         TaskRemoteFilter(String cacheName) {
             this.cacheName = cacheName;
         }

         /** {@inheritDoc} */
         @Override public boolean apply(CacheEvent evt) {
             Affinity<Object> affinity = ignite.affinity(cacheName);

             if (affinity.isPrimary(ignite.cluster().localNode(), evt.key())) {
                 // Process this event. Ignored on backups.
                 if (filter != null && filter.apply(evt))
                     return false;

                 return true;
             }

             return false;
         }
     }

     /**
      * Grid instance initialized on demand.
      */
     private static class IgniteGrid {
         /** Constructor. */
         private IgniteGrid() {
             // No-op.
         }

         /** Instance holder. */
         private static class Holder {
             /** */
             private static final Ignite IGNITE = Ignition.start(igniteCfgFile);
         }

         /**
          * Obtains grid instance.
          *
          * @return Grid instance.
          */
         private static Ignite getIgnite() {
             return Holder.IGNITE;
         }
     }

     /** Cache events available for listening. */
     private enum CacheEvt {
         /** */
         CREATED(EventType.EVT_CACHE_ENTRY_CREATED),
         /** */
         DESTROYED(EventType.EVT_CACHE_ENTRY_DESTROYED),
         /** */
         PUT(EventType.EVT_CACHE_OBJECT_PUT),
         /** */
         READ(EventType.EVT_CACHE_OBJECT_READ),
         /** */
         REMOVED(EventType.EVT_CACHE_OBJECT_REMOVED),
         /** */
         LOCKED(EventType.EVT_CACHE_OBJECT_LOCKED),
         /** */
         UNLOCKED(EventType.EVT_CACHE_OBJECT_UNLOCKED),
         /** */
         EXPIRED(EventType.EVT_CACHE_OBJECT_EXPIRED);

         /** Internal Ignite event id. */
         private final int id;

         /**
          * Constructor.
          *
          * @param id Internal Ignite event id.
          */
         CacheEvt(int id) {
             this.id = id;
         }

         /**
          * Gets Ignite event id.
          *
          * @return Ignite event id.
          */
         int getId() {
             return id;
         }
     }
 }
	/*
	* 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.ignite.stream.kafka.connect;

	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.List;
	import java.util.Map;
	import java.util.UUID;
	import java.util.concurrent.BlockingQueue;
	import java.util.concurrent.LinkedBlockingQueue;
	import java.util.concurrent.TimeUnit;
	import org.apache.ignite.Ignite;
	import org.apache.ignite.IgniteException;
	import org.apache.ignite.Ignition;
	import org.apache.ignite.cache.affinity.Affinity;
	import org.apache.ignite.events.CacheEvent;
	import org.apache.ignite.events.EventType;
	import org.apache.ignite.lang.IgniteBiPredicate;
	import org.apache.ignite.lang.IgnitePredicate;
	import org.apache.ignite.resources.IgniteInstanceResource;
	import org.apache.kafka.connect.errors.ConnectException;
	import org.apache.kafka.connect.source.SourceRecord;
	import org.apache.kafka.connect.source.SourceTask;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* Task to consume remote cluster cache events from the grid and inject them into Kafka.
	* <p>
	* Note that a task will create a bounded queue in the grid for more reliable data transfer.
	* Queue size can be changed by {@link IgniteSourceConstants#INTL_BUF_SIZE}.
	*/
	public class IgniteSourceTask extends SourceTask {
	/** Logger. */
	private static final Logger log = LoggerFactory.getLogger(IgniteSourceTask.class);

	/** Tasks static monitor. */
	private static final Object lock = new Object();

	/** Event buffer size. */
	private static int evtBufSize = 100000;

	/** Event buffer. */
	private static BlockingQueue<CacheEvent> evtBuf = new LinkedBlockingQueue<>(evtBufSize);

	/** Max number of events taken from the buffer at once. */
	private static int evtBatchSize = 100;

	/** Flag for stopped state. */
	private static volatile boolean stopped = true;

	/** Ignite grid configuration file. */
	private static String igniteCfgFile;

	/** Cache name. */
	private static String cacheName;

	/** Remote Listener id. */
	private static UUID rmtLsnrId;

	/** Local listener. */
	private static TaskLocalListener locLsnr = new TaskLocalListener();

	/** User-defined filter. */
	private static IgnitePredicate<CacheEvent> filter;

	/** Topic. */
	private static String topics[];

	/** Offset. */
	private static final Map<String, Long> offset = Collections.singletonMap("offset", 0L);

	/** Partition. */
	private static final Map<String, String> srcPartition = Collections.singletonMap("cache", null);

	/** {@inheritDoc} */
	@Override public String version() {
	return new IgniteSinkConnector().version();
	}

	/**
	* Filtering is done remotely. Local listener buffers data for injection into Kafka.
	*
	* @param props Task properties.
	*/
	@Override public void start(Map<String, String> props) {
	synchronized (lock) {
	// Each task has the same parameters -- avoid setting more than once.
	// Nothing to do if the task has been already started.
	if (!stopped)
	return;

	cacheName = props.get(IgniteSourceConstants.CACHE_NAME);
	igniteCfgFile = props.get(IgniteSourceConstants.CACHE_CFG_PATH);
	topics = props.get(IgniteSourceConstants.TOPIC_NAMES).split("\\s,\\s");

	if (props.containsKey(IgniteSourceConstants.INTL_BUF_SIZE))
	evtBufSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BUF_SIZE));

	if (props.containsKey(IgniteSourceConstants.INTL_BATCH_SIZE))
	evtBatchSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BATCH_SIZE));

	if (props.containsKey(IgniteSourceConstants.CACHE_FILTER_CLASS)) {
	String filterCls = props.get(IgniteSourceConstants.CACHE_FILTER_CLASS);
	if (filterCls != null && !filterCls.isEmpty()) {
	try {
	Class<? extends IgnitePredicate<CacheEvent>> clazz =
	(Class<? extends IgnitePredicate<CacheEvent>>)Class.forName(filterCls);

	filter = clazz.newInstance();
	}
	catch (Exception e) {
	log.error("Failed to instantiate the provided filter! " +
	"User-enabled filtering is ignored!", e);
	}
	}
	}

	TaskRemoteFilter rmtLsnr = new TaskRemoteFilter(cacheName);

	try {
	int[] evts = cacheEvents(props.get(IgniteSourceConstants.CACHE_EVENTS));

	rmtLsnrId = IgniteGrid.getIgnite().events(IgniteGrid.getIgnite().cluster().forCacheNodes(cacheName))
	.remoteListen(locLsnr, rmtLsnr, evts);
	}
	catch (Exception e) {
	log.error("Failed to register event listener!", e);

	throw new ConnectException(e);
	}
	finally {
	stopped = false;
	}
	}
	}

	/** {@inheritDoc} */
	@Override public List<SourceRecord> poll() throws InterruptedException {
	ArrayList<SourceRecord> records = new ArrayList<>(evtBatchSize);
	ArrayList<CacheEvent> evts = new ArrayList<>(evtBatchSize);

	if (stopped)
	return records;

	try {
	if (evtBuf.drainTo(evts, evtBatchSize) > 0) {
	for (CacheEvent evt : evts) {
	// schema and keys are ignored.
	for (String topic : topics)
	records.add(new SourceRecord(srcPartition, offset, topic, null, evt));
	}

	return records;
	}
	}
	catch (IgniteException e) {
	log.error("Error when polling event queue!", e);
	}

	// for shutdown.
	return null;
	}

	/**
	* Converts comma-delimited cache events strings to Ignite internal representation.
	*
	* @param evtPropsStr Comma-delimited cache event names.
	* @return Ignite internal representation of cache events to be registered with the remote listener.
	* @throws Exception If error.
	*/
	private int[] cacheEvents(String evtPropsStr) throws Exception {
	String[] evtStr = evtPropsStr.split("\\s,\\s");

	if (evtStr.length == 0)
	return EventType.EVTS_CACHE;

	int[] evts = new int[evtStr.length];

	try {
	for (int i = 0; i < evtStr.length; i++)
	evts[i] = CacheEvt.valueOf(evtStr[i].toUpperCase()).getId();
	}
	catch (Exception e) {
	log.error("Failed to recognize the provided cache event!", e);

	throw new Exception(e);
	}
	return evts;
	}

	/**
	* Stops the grid client.
	*/
	@Override public synchronized void stop() {
	if (stopped)
	return;

	stopped = true;

	stopRemoteListen();

	IgniteGrid.getIgnite().close();
	}

	/**
	* Stops the remote listener.
	*/
	protected void stopRemoteListen() {
	if (rmtLsnrId != null)
	IgniteGrid.getIgnite().events(IgniteGrid.getIgnite().cluster().forCacheNodes(cacheName))
	.stopRemoteListen(rmtLsnrId);

	rmtLsnrId = null;
	}

	/**
	* Used by unit test to avoid restart node and valid state of the <code>stopped</code> flag.
	*
	* @param stopped Stopped flag.
	*/
	protected static void setStopped(boolean stopped) {
	IgniteSourceTask.stopped = stopped;
	}

	/**
	* Local listener buffering cache events to be further sent to Kafka.
	*/
	private static class TaskLocalListener implements IgniteBiPredicate<UUID, CacheEvent> {
	/** {@inheritDoc} */
	@Override public boolean apply(UUID id, CacheEvent evt) {
	try {
	if (!evtBuf.offer(evt, 10, TimeUnit.MILLISECONDS))
	log.error("Failed to buffer event {}", evt.name());
	}
	catch (InterruptedException e) {
	e.printStackTrace();
	}

	return true;
	}
	}

	/**
	* Remote filter.
	*/
	private static class TaskRemoteFilter implements IgnitePredicate<CacheEvent> {
	/** */
	@IgniteInstanceResource
	Ignite ignite;

	/** Cache name. */
	private final String cacheName;

	/**
	* @param cacheName Cache name.
	*/
	TaskRemoteFilter(String cacheName) {
	this.cacheName = cacheName;
	}

	/** {@inheritDoc} */
	@Override public boolean apply(CacheEvent evt) {
	Affinity<Object> affinity = ignite.affinity(cacheName);

	if (affinity.isPrimary(ignite.cluster().localNode(), evt.key())) {
	// Process this event. Ignored on backups.
	if (filter != null && filter.apply(evt))
	return false;

	return true;
	}

	return false;
	}
	}

	/**
	* Grid instance initialized on demand.
	*/
	private static class IgniteGrid {
	/** Constructor. */
	private IgniteGrid() {
	// No-op.
	}

	/** Instance holder. */
	private static class Holder {
	/** */
	private static final Ignite IGNITE = Ignition.start(igniteCfgFile);
	}

	/**
	* Obtains grid instance.
	*
	* @return Grid instance.
	*/
	private static Ignite getIgnite() {
	return Holder.IGNITE;
	}
	}

	/** Cache events available for listening. */
	private enum CacheEvt {
	/** */
	CREATED(EventType.EVT_CACHE_ENTRY_CREATED),
	/** */
	DESTROYED(EventType.EVT_CACHE_ENTRY_DESTROYED),
	/** */
	PUT(EventType.EVT_CACHE_OBJECT_PUT),
	/** */
	READ(EventType.EVT_CACHE_OBJECT_READ),
	/** */
	REMOVED(EventType.EVT_CACHE_OBJECT_REMOVED),
	/** */
	LOCKED(EventType.EVT_CACHE_OBJECT_LOCKED),
	/** */
	UNLOCKED(EventType.EVT_CACHE_OBJECT_UNLOCKED),
	/** */
	EXPIRED(EventType.EVT_CACHE_OBJECT_EXPIRED);

	/** Internal Ignite event id. */
	private final int id;

	/**
	* Constructor.
	*
	* @param id Internal Ignite event id.
	*/
	CacheEvt(int id) {
	this.id = id;
	}

	/**
	* Gets Ignite event id.
	*
	* @return Ignite event id.
	*/
	int getId() {
	return id;
	}
	}
	}