Aws Kinesis Integration (sending messages to kinesis)

I will be sharing here details about integration of Aws Kinesis here. If you have bit time, I would recommend you to visit here


1. Create a properties file for Kinesis.

@Component
public class KinesisProp {

    @Value("${kinesis.access.key}")
    private String  awsAccessKey;
    @Value("${kinesis.secret.key}")
    private String  awsSecretKey;
    @Value("${kinesis.streamName}")
    private String streamName;
    @Value("${kinesis.receive.limit}")
    private Integer receiveLimit;
    @Value("${kinesis.shradCount}")
    private Integer shardCount;
    @Value("${kinesis.region}")
    private String region;
    @Value("${kinesis.steam.initialPosition}")
    private String streamInitialPosition;
    @Value("${kinesis.endPoint}")
    private String endPoint; 

    // getters and setters here

}

2. Create a service call, which will be responsible for sending messages to Kinesis.

@Component
public class KinesisService {
    private static final Logger logger = LoggerFactory.getLogger(KinesisService.class);
    @Autowired
    private KinesisProp kinesisProp;
    private AmazonKinesisClient kinesisClient;
 
    @PostConstruct
    private void init() {
        try {
            instantiateKinesisClient();
        } catch (Exception e) {
            logger.error("Exception occurred while instantiating the Kinesis Client", e);
        }
    }

    private void instantiateKinesisClient() throws IOException {
        BasicAWSCredentials  credentials = new BasicAWSCredentials(this.kinesisProp.getAwsAccessKey(), this.kinesisProp.getAwsSecretKey());
        kinesisClient = new AmazonKinesisClient(credentials);
        final Region apac_east = Region.getRegion(Regions.valueOf(kinesisProp.getRegion()));
        kinesisClient.setRegion(apac_east);
        createStream();
    }

    private void createStream() {
        final ListStreamsRequest listStreamsRequest = new ListStreamsRequest();
        ListStreamsResult listStreamsResult = kinesisClient.listStreams(listStreamsRequest);
        if (listStreamsResult == null || listStreamsResult.getStreamNames().isEmpty() ||
                (!listStreamsResult.getStreamNames().contains(kinesisProp.getStreamName()))) {

            CreateStreamRequest createStreamRequest = new CreateStreamRequest();
            createStreamRequest.setStreamName(kinesisProp.getStreamName());
            createStreamRequest.setShardCount(kinesisProp.getShardCount());

            try {
                kinesisClient.createStream(createStreamRequest);
            } catch (Exception e) {
                logger.error("Error occurred while creating the stream!", e);
            }
        }
    }

    public void sendMessage(final String message) throws AmazonServiceException {
       PutRecordRequest putRecordRequest = new PutRecordRequest();
 putRecordRequest.setStreamName(kinesisProp.getStreamName());
 putRecordRequest.setData(ByteBuffer.wrap(message.getBytes()));                                                                      putRecordRequest.setPartitionKey(Constants.AWS_KINESIS_PARTITION_KEY);
          kinesisClient.putRecord(putRecordRequest);
    }
 
}


3. Fetching the messages can be done in two ways.

(a) Using Aws Kinesis API
(b) Using Aws Kinesis Client Library.

I will update on this in my next blog.

Note:
(a) Spring is used for DI.
(b) When you are creating the access key and secret key, make sure to create for the kinesis app.

Embedded jetty server with web app

Recently, I come across a situation where both maven-jetty plugin & embedded jetty server need to packed along with web-application. I have already written a blog on incorporating maven-jetty plugin in the web app here, hence will not be discussing here again. I will describe here about package of embedded jetty server with in web application below.

Firstly, why do we require to include embedded jetty?
1. Easy deployment, java -jar <web-app>.jar, once you build the artifact, it's ready for deployment without any hassle configurations/deployment procedures, etc
2. No requirement of application server.


Let's go through the following steps to include jetty server within web app.

1. Create a JettyServer class


package com.sateesh.application.server;

import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.servlet.DefaultServlet;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.webapp.WebAppContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.servlet.DispatcherServlet;

import java.io.IOException;


public class JettyServer {

    private static final Logger logger = LoggerFactory.getLogger(JettyServer.class);
    private final String WEB_APP_DIR = "src/main/webapp";
    private final String WEB_XML_PATH = "src/main/webapp/WEB-INF/web.xml";
    private static final String MVC_SERVLET_NAME = "dispatcher";
    private static final String JSP_SERVLET_NAME = "jspServlet";
    private static final String DEFAULT_PORT = "8080";
    private static final String DEFAULT_CONTEXT_PATH = "/";



    public static void main(String[] args) throws Exception{
        JettyServer jettyServer = new JettyServer();
        jettyServer.start();
    }

    private JettyServer() {
      // none can instantiate this class
    }

    private void start() throws Exception{
        try {
            QueuedThreadPool threadPool = new QueuedThreadPool(512);

            final Server server = new Server(threadPool);
            server.setStopAtShutdown(true);
            server.setStopTimeout(5000);

            ServerConnector connector = new ServerConnector(server);
            connector.setPort(getPort());
            connector.setIdleTimeout(30000);
            server.setConnectors(new Connector[]{connector});
            server.setHandler(getServletContextHandler());

            server.start();
            server.join();
            logger.info("Embedded Jetty Server is started!");
        } catch (Exception e) {
            logger.error("Exception occurred while starting the jetty server ", e);
        }
    }

    private WebAppContext getServletContextHandler() throws IOException {
        final WebAppContext contextHandler = new WebAppContext();
        contextHandler.setContextPath(getContextPath());
        contextHandler.setWar(WEB_APP_DIR);
        contextHandler.setDefaultsDescriptor(WEB_XML_PATH);
        ServletHolder jspServletHolder = new ServletHolder("jspServlet", new org.apache.jasper.servlet.JspServlet());
        // jsp servelet is required to parse the jsp passed by the spring-mvc  
        contextHandler.addServlet(jspServletHolder, "*.jsp");
        return contextHandler;
    }

 
  /**
  * Consider default port as "8080" until unless not passed using java args
  **/  private int getPort() {
        Integer port = Integer.parseInt(System.getProperty("jetty.port"));

        if (port == null || port == 0) {
            port = Integer.parseInt(DEFAULT_PORT);
        }

        return port;
    }

    
  /**
  * Consider default context path as "/" until unless not passed using java args
  **/
  private String getContextPath() {
        String contextPath = System.getProperty("jetty.contextPath");

        if (contextPath == null ||  contextPath.equals("")) {
            contextPath = DEFAULT_CONTEXT_PATH;
        }

        return contextPath;
    }

}


2. Contents of web.xml (default descriptor executed by JettyServer).

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE web-app PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN"
        "http://java.sun.com/dtd/web-app_2_3.dtd">
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns="http://java.sun.com/xml/ns/javaee"
        xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
        id="YourWebAppID"
        version="2.5">

    <display-name>Web App Server</display-name>

    <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>

    <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>
            classpath:META-INF/spring/application-context.xml
        </param-value>
    </context-param>

    <servlet>
        <servlet-name>dispatcher</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:dispatcher-servlet.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
    </servlet>

    <servlet-mapping>
        <servlet-name>dispatcher</servlet-name>
        <url-pattern>/api/*</url-pattern>
    </servlet-mapping>

</web-app>


3. Contents of application-context.xml & dispatcher-servlet.xml files

application-context.xml --- is spring descriptor file. 
dispatcher-servlet.xml --- is servlet dispatcher file, which will scan @Controller annotated packages.


4. Package structure of the project

   --- src
       --- main
           --- java 
               --- com.sateesh.application.server
           --- resources
               --- META-INF
                   --- spring
                       --- application-context.xml
               --- dispatcher-servlet.xml
               --- logback.xml
           --- webapp
               --- views
                   ---  ***.jsp
               --- web.xml


5. below plugin needs to be added in pom.xml

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <createDependencyReducedPom>true</createDependencyReducedPom>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.sateesh.application.server.JettyServer</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>

6. add below dependencies in pom.xml

    <properties>
        <jetty.server.scope>compile</jetty.server.scope>
        <jetty.version>9.1.2.v20140210</jetty.version>
    </properties>

        <dependency>
            <groupId>org.eclipse.jetty</groupId>
            <artifactId>jetty-server</artifactId>
            <version>${jetty.version}</version>
            <scope>${jetty.server.scope}</scope>
        </dependency>
        <dependency>
            <groupId>org.eclipse.jetty</groupId>
            <artifactId>jetty-webapp</artifactId>
            <version>${jetty.version}</version>
            <scope>${jetty.server.scope}</scope>
        </dependency>
        <dependency>
            <groupId>org.eclipse.jetty</groupId>
            <artifactId>jetty-servlet</artifactId>
            <version>${jetty.version}</version>
            <scope>${jetty.server.scope}</scope>
        </dependency>
        <dependency>
            <groupId>org.eclipse.jetty</groupId>
            <artifactId>jetty-jsp</artifactId>
            <version>${jetty.version}</version>
            <scope>${jetty.server.scope}</scope>
        </dependency>

7. You are almost there.. now run the web application now.

build the project (default package is jar in maven):
mvn clean package

run the webapp:
java -jar target/webapp.jar


PS: we can also create embedded jetty server with in web-app without web.xml file also.