Showing posts with label simple. Show all posts
Showing posts with label simple. Show all posts

Wednesday, June 13, 2018

Oauth2 with Spring Boot simple example

I this post, using spring boot, I'll show a basic Oauth2 flow with :
 - Authorization server
 - Client app which logs in to Authorization server using username and password, takes login token as a response of successful login and calls resource server with received token.
 - Resource server(which have protected resource) handles requests, grabs token from the request, validates tokens on Authorization server, returns requested data.

So,  I have to create 3 separate spring boot applications(Authorization server, Client app, Resource server), run them, and make sure all flow works.


0. Dependencies

For simplicity I'm using the same dependencies for all 3 applications:
dependencies {
    compile('org.springframework.boot:spring-boot-starter-web')
    compile('org.springframework.cloud:spring-cloud-starter-oauth2')
    compile('org.springframework.cloud:spring-cloud-starter-security')
    testCompile('org.springframework.boot:spring-boot-starter-test')
}

1. Authorization server

Authorization server it's a spring boot application which will be used to authorize user by credentials sent by client application. As a response it should send a token back to client. 
In properties, we are defining clientId, clientSecret (password) which should be used for authorization. Also we should define grant-types, and if we need them - scopes (in this example scope will not be used).

application.properties:

server.port: 9000
server.servlet.context-path: /servicessecurity.oauth2.client.clientId: myClientIdsecurity.oauth2.client.clientSecret: myClientSecretsecurity.oauth2.client.authorized-grant-types: authorization_code,refresh_token,password,client_credentialssecurity.oauth2.client.scope:data_insert,data_update, data_delete, data_select


In service config file  I' defining users(huan and joe) of my application with passwords. Of course, in real life they will not be defined in a code, but should be stored in DB.

ServiceConfig:
package com.demien.sboot.oauthserver;
import org.springframework.context.annotation.Configuration;import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder;import org.springframework.security.config.annotation.authentication.configuration.GlobalAuthenticationConfigurerAdapter;
@Configurationpublic class ServiceConfig extends GlobalAuthenticationConfigurerAdapter {
    @Override    public void init(AuthenticationManagerBuilder auth) throws Exception {
        auth.inMemoryAuthentication()
                .withUser("huan").password("{noop}sebastyan").roles("USER") .and()
                .withUser("joe").password("{noop}black").roles("USER", "ADMIN");    }
}


In server runner application we have to define endpoint for getting user details - we will use it later.

Server runner application:
package com.demien.sboot.oauthserver;
import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.security.oauth2.config.annotation.web.configuration.EnableAuthorizationServer;import org.springframework.security.oauth2.config.annotation.web.configuration.EnableResourceServer;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;
import java.security.Principal;
@SpringBootApplication@EnableAuthorizationServer@EnableResourceServer@RestControllerpublic class OauthServerApp {
    public static void main(String[] args) {
        SpringApplication.run(OauthServerApp.class, args);    }

    @RequestMapping("/user")
    public Principal user(Principal user) {
        return user;    }
}


2. Resource server

Resource server is a spring boot application which has some protected resource(endpoint "/mydata"), which is not accessible without authorization.  Client should provide authorization token to call this endpoint. In properties file we should define endpoint from authorization server mentioned above.

application.properties
server.port=9001server.servlet.context-path=/servicessecurity.oauth2.resource.userInfoUri:http://localhost:9000/services/user


Service config:

package com.demien.sboot.service;
import org.springframework.context.annotation.Configuration;import org.springframework.security.access.expression.method.MethodSecurityExpressionHandler;import org.springframework.security.config.annotation.method.configuration.EnableGlobalMethodSecurity;import org.springframework.security.config.annotation.method.configuration.GlobalMethodSecurityConfiguration;import org.springframework.security.oauth2.provider.expression.OAuth2MethodSecurityExpressionHandler;
@Configuration@EnableGlobalMethodSecurity(prePostEnabled = true)
public class ServiceConfig extends GlobalMethodSecurityConfiguration {

    @Override    protected MethodSecurityExpressionHandler createExpressionHandler() {
        return new OAuth2MethodSecurityExpressionHandler();    }
}


In main class we're just defining endpoint and class with some very important data.

Application runner:

package com.demien.sboot.service;
import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.boot.autoconfigure.security.oauth2.resource.ResourceServerProperties;import org.springframework.context.annotation.Bean;import org.springframework.security.access.prepost.PreAuthorize;import org.springframework.security.oauth2.config.annotation.web.configuration.EnableResourceServer;import org.springframework.security.oauth2.provider.token.ResourceServerTokenServices;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;
import java.util.ArrayList;
@SpringBootApplication@RestController@EnableResourceServerpublic class ServiceApp {

    public static void main(String[] args) {
        SpringApplication.run(ServiceApp.class, args);    }

    @RequestMapping("/mydata")
    public ArrayList<MyData> getTollData() {

        ArrayList<MyData> result = new ArrayList<MyData>();        result.add(new MyData(1L, "one"));        result.add(new MyData(2L, "two"));        result.add(new MyData(3L, "three"));
        return result;    }


    public class MyData {

        public final Long myId;        public final String myValue;
        public MyData(Long myId, String myValue) {
            this.myId = myId;            this.myValue = myValue;        }

        public Long getMyId() {
            return myId;        }

        public String getMyValue() {
            return myValue;        }
    }


}

3. Command line client 


Client application is a most interesting thing here. First of all we have to define in properties detail for authorization.

application.yml
server:  port: 9090
  servlet:    context-path: /services

security:  oauth2:    client:      clientId: myClientId
      clientSecret: myClientSecret
      accessTokenUri: http://localhost:9000/services/oauth/token
      userAuthorizationUri: http://localhost:9000/services/oauth/authorize
      clientAuthenticationScheme: form
    resource:      userInfoUri: http://localhost:9000/services/user
      preferTokenInfo: false


In main application runner we are about to call our protected endpoint: http://localhost:9001/services/mydata
But for this call we should be authorized first. I'm using credentials of user "joe" for this.
Also I'm printing authorization token, to make sure we have it.
And finally I'm calling this endpoint. 

Application runner:
package com.demien.sboot.client;
import org.springframework.boot.CommandLineRunner;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.security.oauth2.client.OAuth2RestTemplate;import org.springframework.security.oauth2.client.token.grant.password.ResourceOwnerPasswordResourceDetails;import org.springframework.security.oauth2.common.AuthenticationScheme;
import java.util.Arrays;
@SpringBootApplicationpublic class CommandLineApp implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(CommandLineApp.class, args);    }


    @Override    public void run(String... args) throws Exception {
        System.out.println("starting");        ResourceOwnerPasswordResourceDetails resourceDetails = new ResourceOwnerPasswordResourceDetails();        resourceDetails.setAuthenticationScheme(AuthenticationScheme.header);        resourceDetails.setAccessTokenUri("http://localhost:9000/services/oauth/token");        resourceDetails.setScope(Arrays.asList("data_select"));        resourceDetails.setClientId("myClientId");        resourceDetails.setClientSecret("myClientSecret");        resourceDetails.setUsername("joe");        resourceDetails.setPassword("black");
        OAuth2RestTemplate restTemplate = new OAuth2RestTemplate(resourceDetails);        String token = restTemplate.getAccessToken().getValue();        System.out.println("token:" + token);
        String s = restTemplate.getForObject("http://localhost:9001/services/mydata", String.class);        System.out.println("Result:" + s);    }
}

4. Execution

First of all we should start Authorization and resource services.
After that, we're free to go: let's start our client.
For me it produced next output:

starting
token:2b9560c9-355d-4134-a63b-e10f05b1b9b4
Result:[{"myId":1,"myValue":"one"},{"myId":2,"myValue":"two"},{"myId":3,"myValue":"three"}]

It looks simple, but under the hood client called authorization server, to get token, called resource service with token saved in session. Resource service called authorization server again to validate the token and after that - returned result back to client.


5. The end

Source code can be downloaded from here

Sunday, March 18, 2018

Python: getting started


Intro

Last time Python is getting more and more popular in BigData word.
From wiki:
Python is an interpreted high-level programming language for general-purpose programming. Created by Guido van Rossum and first released in 1991, Python has a design philosophy that emphasizes code readability, and a syntax that allows programmers to express concepts in fewer lines of code,[26][27] notably using significant whitespace. It provides constructs that enable clear programming on both small and large scales.[28]
Python features a dynamic type system and automatic memory management. It supports multiple programming paradigms, including object-orientedimperativefunctional and procedural, and has a large and comprehensive standard library.[29]
Python interpreters are available for many operating systemsCPython, the reference implementation of Python, is open sourcesoftware[30] and has a community-based development model, as do nearly all of its variant implementations. CPython is managed by the non-profit Python Software Foundation.

From official documentation:
Python is powerful... and fast; 
plays well with others; 
runs everywhere; 
is friendly & easy to learn; 
is Open.


Installation

Installation process may vary depending on operational system: it can be installation on windows or just execution of "apt-get" in linux, but anyway, on official site you can find any information you need. 

Python shell

Easiest way to play with python - command shell, which can be open by running "python" command:

demien$ python
Python 2.7.13 (default, Nov 24 2017, 17:33:09)
[GCC 6.3.0 20170516] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>>

Now we can output standard "hello world" greeting message: 
>>> print("hello world")
hello world


Variables can be created by simple "=" operator: 
>>> name = "Huan Sebastyan"
>>> print("hello, {0} !!!".format(name))
hello, Huan Sebastyan !!!


PY Files

For long programs, python shell is not an option: it's better to put program content into file(or files) with extension ".py"  and run it by executing: python myprogram.py.  In code examples below I'm writing the code into file test.py  and executing it by running: python test.py

The main issue with Python files are the "blocks" of code : Python doesn't have {} or begin end operators to define scope(beginning and ending) of block, function or class. For this purpose Python use spaces: 

operator1
operator2 
def myFunction:
    function operator1
    function operator2
operator 3


Conditions in Python

Conditions are written without any bracers and of course for block inside condition we should use spaces to define begin and end:  

name = raw_input("What is your name? ")
greeting =""
if name == "Huan Sebastyan":
greeting = "Buenos dias"
else:
greeting = "Hello"
print("{0}, {1} !!!".format(greeting, name))

execution:

demien$ python test.py
What is your name? Joe
Hello, Joe !!!

demien$ python test.py
What is your name? Huan Sebastyan
Buenos dias, Huan Sebastyan !!!

Loops

There several types of loops in python: while and for

While loop:


value=""
while value!="end":
value = raw_input("Enter 'end' to quit or anything else to continue: ")
print("you entered:{0}".format(value))
print("you did it!")


execution: 
demien$ python test.py
Enter 'end' to quit or anything else to continue: hello
you entered:hello
Enter 'end' to quit or anything else to continue: world
you entered:world
Enter 'end' to quit or anything else to continue: end
you entered:end
you did it!

For loop:

count = int(raw_input("enter iteration count: "))
for i in range(1, count+1):
s = ""
for j in range(1, i+1):
s+="*"
print(s)

execution:
demien$ python test.py
enter iteration count: 5
*
**
***
****
*****


Functions

Definition begins with keyword "def" , bracers used for parameters: 


x = 10
y = 20

def test(param):
global x
y = 20
x=x+1
y=y+1
print("from function context: param={0}, x={1}, y={2}".format(param, x, y))
return "BYE!"

result = test("HELLO!")
print("from global context: result={0}, x={1}, y={2}".format(result, x, y))

Execution:
demien$ python test.py
from function context: param=HELLO!, x=11, y=21
from global context: result=BYE!, x=11, y=20


Modules

For big programs it's impossible to keep all code in one file. Python provides concept MODULE for this: 
Let's create a simple module in subfolder "tools":

tools/simple.py: 

def sayHi(name):
print("Hi, {0}".format(name))
__version__ = "0.0.1"

If we want to use it, beside module itself we also need an empty file __init__.py in the same folder(in subfolder "tools"):
demien$ ls -la tools/*.py
-rw-r--r-- 1 demien demien  0 mar 11 14:49 tools/__init__.py
-rw-r--r-- 1 demien demien 78 mar 11 14:46 tools/simple.py

Now we can use this module from our main program: 

import tools.simple as simple

simple.sayHi("Joe")


Execution: 
demien$ python test.py
Hi, Joe


Sometimes we may need to understand if the code is running by "direct" execution, or by importing it as module. For this we can check condition: if __name__ == "__main__":
- if it returns true, the code is being executed in "direct" mode. Our current module by direct execution produce no output. Let's update our module to produce some output, but only in case of "direct" execution: 

def sayHi(name):
print("Hi, {0}".format(name))
__version__ = "0.0.1"
if __name__ == "__main__":
sayHi("Huan Sebastyan")


So, now we can run our module in a direct way: 

demien$ python tools/simple.py
Hi, Huan Sebastyan

And now let's use it as module: 

demien$ python test.py
Hi, Joe

- output remained the same. 


Dir function

This function is showing "content" of variables and methods defined in module: 

>>> import tools.simple
>>> dir(tools.simple)
['__builtins__', '__doc__', '__file__', '__name__', '__package__', '__version__', 'sayHi']


Collection classes

there several collection classes in Python: lists, dictionary, set, tuples

Lists

It's a mutable structures for storing data in arrays.

colors = ["red", "black", "white"]
print "there are ",len(colors)," colors in my list:"
for color in colors:
print(color)

colors.append("blue")
colors.append("green")
print "few more there added, now it's ",len(colors)," of them"

colors.sort()
print("sorted:", colors)

del colors[0]
del colors[0]
print("first 2 were deleted:", colors)

Execution: 
demien$ python test.py
there are  3  colors in my list:
red
black
white
few more there added, now it's  5  of them
('sorted:', ['black', 'blue', 'green', 'red', 'white'])
('first 2 were deleted:', ['green', 'red', 'white'])

Tuples

Similar to lists  but they are immutable

answer = ("yes", "no")
print answer[0]
print answer[1]
execution:
demien$ python test.py
yes
no


Dictionary

It's structure like Map or Associated array. 

user = {
"name" : "Joe",
"surname" : "Black",
"address" : {
"country" : "USA",
"city" : "Houston"
}
}

print user["name"]
print user["address"]
print user["address"]["city"]

print "full list of pairs[key,value] in user dictionary:"
for key, value in user.items():
print "key=", key, " value=", value


Execution:
demien$ python test.py
Joe
{'country': 'USA', 'city': 'Houston'}
Houston
full list of pairs[key,value] in user dictionary:
key= surname  value= Black
key= name  value= Joe
key= address  value= {'country': 'USA', 'city': 'Houston'}


Sequence operations. 

Structures like lists, tuples and strings have list of common "sequence" operations.

colors = ["red", "black", "white", "blue", "gray", "green", "orange"]
colors.sort()
print colors

print('color 2 is', colors[2])
print('color -2 is', colors[-2])

print('colors 1 to 3 is', colors[1:3])
print('colors 2 to end is', colors[2:])
print('colors 1 to -1 is', colors[1:-1])
print('colors start to end is', colors[:])

Execution:
demien$ python test.py
['black', 'blue', 'gray', 'green', 'orange', 'red', 'white']
('color 2 is', 'gray')
('color -2 is', 'red')
('colors 1 to 3 is', ['blue', 'gray'])
('colors 2 to end is', ['gray', 'green', 'orange', 'red', 'white'])
('colors 1 to -1 is', ['blue', 'gray', 'green', 'orange', 'red'])
('colors start to end is', ['black', 'blue', 'gray', 'green', 'orange', 'red', 'white'])


Set

Another collection structure is Set. On sets we can apply some math logic, like AND, OR, XOR: 

colors1 = set(["red", "black", "white"])
colors2 = set(["white", "blue", "gray"])
print(colors1)
print(colors2)

print("& : ", colors1 & colors2)
print("| : ", colors1 | colors2)
print("^ : ", colors1 ^ colors2)
Execution: 
demien$ python test.py
set(['white', 'black', 'red'])
set(['blue', 'gray', 'white'])
('& : ', set(['white']))
('| : ', set(['blue', 'gray', 'black', 'white', 'red']))
('^ : ', set(['blue', 'gray', 'black', 'red']))



Classes

Main issues with  OOP in Python are:
- classes are being created without "new" keyword:  userJoe = User("Joe", "Black")
 - constructor has name: __init__
 - variables defined in class definition are class(not object) variables and should be accessed by className.variableName
 - "self" stands for "this" variable, and used for definition of object variables: self.name = name
-  methods which are using object variables have to explicitly define self as first input parameter: def sayHi(self)
- to define subclass, superclass name has to be passed as a "parameter" for subclass name: class AdminUser(User)
- to call superclass method (even constructor) format should be SuperClassName.SuperClassMethod: User.__init__(self, name, surname)


Example: 

class User:
userCount = 0

def __init__(self, name, surname):
self.name = name
self.surname = surname
User.userCount+=1
print("User #{0} was created!".format(User.userCount))

def sayHi(self):
print("Hi, I'm {0} {1}".format(self.name, self.surname) )

class AdminUser(User):
def __init__(self, name, surname, role):
User.__init__(self, name, surname)
self.role = role

def sayHi(self):
User.sayHi(self)
print(" and I'm the {0} !!!".format(self.role))

userJoe = User("Joe", "Black")
userJoe.sayHi()

userHuan = User("Huan", "Seastyan")
userHuan.sayHi()

userAdmin = AdminUser("Super", "Admin", "boss")
userAdmin.sayHi()
Execution: 
demien$ python test.py
User #1 was created!
Hi, I'm Joe Black
User #2 was created!
Hi, I'm Huan Seastyan
User #3 was created!
Hi, I'm Super Admin
  and I'm the boss !!!

Files

For working with files all we need is "file()" operation which takes parameters : 
1. file name
2. mode: read or write 

Let's create simple file in.txt with 2 lines of text: 
it's a test file
just as example

Now let's create a simple program which will convert this file to uppercase and write it into "out.txt" file: 

inFile = file("in.txt", "r")
outFile = file("out.txt", "w")

eof = False
while eof == False :
line = inFile.readline()
if len(line)==0:
eof = True
else:
outFile.write(line.upper())

inFile.close()
outFile.close()


Execution result
out.txt:
IT'S A TEST FILE
JUST AS EXAMPLE

Pickle

Module pickele (which should be imported) provides ability to save the object into file in "serialized" format. And, of course, later we can read it and deserialize back to original object. 

import pickle

class User:
userCount = 0

def __init__(self, name, surname):
self.name = name
self.surname = surname
User.userCount+=1
print("User #{0} was created!".format(User.userCount))

def sayHi(self):
print("Hi, I'm {0} {1}".format(self.name, self.surname) )

userJoe = User("Joe", "Black")

fout = open("joe.bak", "wb")
pickle.dump(userJoe, fout)
fout.close()
del(userJoe)

fin = open("joe.bak", "rb")
restored = pickle.load(fin)
print(restored)
restored.sayHi()
fin.close()

Execution: 
demien$ python test.py
User #1 was created!
<__main__.User instance at 0x7f05c0296ef0>
Hi, I'm Joe Black

Exceptions

Python has similar to other languages system of error handling with TRY, EXCEPT (which stands for CATCH) and FINALLY. On next example we are handling keyboard input  exceptions such as pressing Ctrl+C during input and raising our own exception if input length is less than expected: 

class ShortInputException(Exception):
"""A user-defined exception class."""

def __init__(self, length, atleast):
Exception.__init__(self)
self.length = length
self.atleast = atleast


try:
text = raw_input("Enter something ....")
if len(text) < 3 :
raise ShortInputException(len(text), 3)

except EOFError:
print("Why did you do an EOF on me?")

except KeyboardInterrupt:
print("You cancelled the operation.")

except ShortInputException as ex:
print(("ShortInputException: The input was " +
"{0} long, expected at least {1}")
.format(ex.length, ex.atleast))

else:
print("You entered {}".format(text))

finally:
print("done")


Execution
demien$ python test.py
Enter something ....12345
You entered 12345
done

demien$ python test.py
Enter something ....12
ShortInputException: The input was 2 long, expected at least 3
done

demien$ python test.py
Enter something ....^CYou cancelled the operation.
done


Try with resource: WITH

If we are opening something in TRY block, very often we have to close it in a FINALLY block (try with resource). To make this "automatically" we can use WITH construction: opened resource will be closed automatically: 

with open("in.txt") as fin:
for line in fin:
print(line)



The end

As for me, python has a lot of common with javascript: dynamic typing, inheritance, but more focused on "back-end" development of scripting. It's very simple but powerful.  Now a lot of big data frameworks are providing python api, so it's better to be familiar with this language. 

Tuesday, January 23, 2018

Apache Kafka - getting started. Simple java project.

0. Intro

Kafka® is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies.

From wiki
The project aims to provide a unified, high-throughput, low-latency platform for handling real-time data feeds. Its storage layer is essentially a "massively scalable pub/sub message queue architected as a distributed transaction log,"[3] making it highly valuable for enterprise infrastructures to process streaming data. Additionally, Kafka connects to external systems (for data import/export) via Kafka Connect and provides Kafka Streams, a Java stream processing library.
The design is heavily influenced by transaction logs.[4]


Last time Apache Kafka is getting more and more popular. With growing popularity of event-sourcing concept, more and more developers are switching to Kafka as primary storage of events. Kafka has everything for this: it's very hast, compact, scalable, "user-friendly"....
In this post I'll show basic simple operations like "send"(by producer) and "receive"(by consumer) messages.


1. Downloading and running kafka

This page is explaining very well how to download and run kafka. If you're using Windows, you can use next commands from "bin/windows" folder:

Run these commands from your Kafka root folder:
cd bin/windows
Then run Zookeper server:
zookeeper-server-start.bat ../../config/zookeeper.properties
Then run Kafka server:
kafka-server-start.bat ../../config/server.properties

Now when kafka is running you can check it by creating a topic and getting topic list:

Create a topic:
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test0
List topics:
kafka-topics.bat --list --zookeeper localhost:2181

Response should be something like:
D:\Projects\kafka_2.11-1.0.0\bin\windows>kafka-topics.bat --list --zookeeper localhost:2181
test0


2. Project structure

Our project structure is very simple: we need just 2 files MessageProducer and MessageConsumer. 

build.gradle file:

group 'com.demien'version '1.0-SNAPSHOT'
apply plugin: 'java'
sourceCompatibility = 1.8
repositories {
    mavenCentral()
}

dependencies {
    compile 'org.apache.kafka:kafka-clients:0.9.0.0'    compile 'org.slf4j:slf4j-api:1.7.12'    compile 'org.slf4j:slf4j-log4j12:1.7.12'    compile 'log4j:log4j:1.2.17'

    testCompile group: 'junit', name: 'junit', version: '4.11'}



3. Producer

It designed as generic by KEY,VALUE types. Also I added to constructor optional messageSentCallback parameter - this callBack will be called when message was sent.


package com.demien.kafka;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;

public class MessageProducer<K, V> {

    private final Producer kafkaProducer;
    private final String topicName;
    private final Consumer<RecordMetadata> messageSentCallback;

    public MessageProducer(String topicName) {
        this(topicName, null);
    }

    public MessageProducer(String topicName, Consumer<RecordMetadata> messageSentCallback) {
        Properties configProperties = new Properties();
        configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        this.kafkaProducer = new KafkaProducer(configProperties);
        this.topicName = topicName;
        this.messageSentCallback = messageSentCallback;
    }


    public void sendMessage(K key, V value) {
        ProducerRecord<K, V> rec = new ProducerRecord<K, V>(topicName, key, value);
        Future<RecordMetadata> future = kafkaProducer.send(rec);
        if (messageSentCallback != null) {
            CompletableFuture.supplyAsync(() -> {
                try {
                    RecordMetadata recordMetadata = future.get();
                    messageSentCallback.accept(recordMetadata);
                } catch (Exception e) {
                }
                return null;
            });
        }
    }

    public void close() {
        kafkaProducer.close();
    }


    public static void main(String[] args) throws InterruptedException {
        MessageProducer<String, String> testProducer = new MessageProducer<String, String>("test0", (recordMetadata) -> {
            System.out.println("Message was sent: offset:" + recordMetadata.offset() + " partition:" + recordMetadata.partition() + " topic:" + recordMetadata.topic());
        });
        testProducer.sendMessage(null, "Test 1 " + new Date().toString());
        testProducer.sendMessage(null, "Test 2 " + new Date().toString());
        testProducer.sendMessage(null, "Test 3 " + new Date().toString());
        testProducer.close();
    }


}



4. Consumer

This class is more complicated, because it's designed to deal with the offsets for reading the data.
Consumer can start reading form the beginning, from the end, or from provided offset. That is why constructor is so complicated. Method for receiving messages is pretty simple. Supplier for cuncumed messages is provided in constructor.

package com.demien.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.Properties;

import java.util.function.BiConsumer;

public class MessageConsumer<K, V> {
    private final String topic;
    private final String groupId;
    private final long startingOffset;
    private final KafkaConsumer<K, V> kafkaConsumer;

    public MessageConsumer(String topic, String groupId) {
        this(topic, groupId, -1);
    }

    /**     * @param topic - id of topic     * @param groupId - id of consumer group     * @param startingOffset - offset to read messages. 0 - from the beginning.      *                       -1 - from the end. other values - start reading from this value                            */    public MessageConsumer(String topic, String groupId, long startingOffset) {
        this.topic = topic;
        this.groupId = groupId;
        this.startingOffset = startingOffset;

        Properties configProperties = new Properties();
        configProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        configProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        configProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        configProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, "testClient");
        configProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        configProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        kafkaConsumer = new KafkaConsumer<>(configProperties);

        kafkaConsumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.printf("%s topic-partitions are revoked from this consumer\n", Arrays.toString(partitions.toArray()));
            }

            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.printf("%s topic-partitions are assigned to this consumer\n", Arrays.toString(partitions.toArray()));
                Iterator<TopicPartition> topicPartitionIterator = partitions.iterator();
                while (topicPartitionIterator.hasNext()) {
                    TopicPartition topicPartition = topicPartitionIterator.next();
                    System.out.println("Current offset is " + kafkaConsumer.position(topicPartition) + " committed offset is ->" + kafkaConsumer.committed(topicPartition));
                    if (MessageConsumer.this.startingOffset == 0) {
                        System.out.println("Setting offset to begining");

                        kafkaConsumer.seekToBeginning(topicPartition);
                    } else if (MessageConsumer.this.startingOffset == -1) {
                        System.out.println("Setting it to the end ");

                        kafkaConsumer.seekToEnd(topicPartition);
                    } else {
                        System.out.println("Resetting offset to " + MessageConsumer.this.startingOffset);
                        kafkaConsumer.seek(topicPartition, MessageConsumer.this.startingOffset);
                    }
                }
            }
        });

    }

    public void startReceiving(BiConsumer<K, V> biConsumer) {
        try {
            while (true) {
                ConsumerRecords<K, V> records = kafkaConsumer.poll(100);
                records.forEach(record->  biConsumer.accept(record.key(), record.value()));
                if (startingOffset == -2) kafkaConsumer.commitSync();
            }
        } finally {
            kafkaConsumer.close();
        }
    }

    public static void main(String[] args) {
        final MessageConsumer<String, String> testConsumer = new MessageConsumer<>("test0", "testGroup");
        testConsumer.startReceiving( (k,v) -> System.out.println("received:"+v) );

    }
}


5. Execution 

Let's start the Consumer now. It should output something like:

[test0-0] topic-partitions are assigned to this consumer
Current offset is 0 committed offset is ->null


Now let's start the Producer. Is should send 3 test messages and print information about them:

Message was sent: offset:0 partition:0 topic:test0
Message was sent: offset:1 partition:0 topic:test0
Message was sent: offset:2 partition:0 topic:test0

Consumer also should print information about received messages:

received:Test 1 Tue Jan 23 15:36:29 CET 2018
received:Test 2 Tue Jan 23 15:36:29 CET 2018
received:Test 3 Tue Jan 23 15:36:29 CET 2018

Let's restart our consumer now. By default value in our constructor, if will be reading data from the end, so previous messages will not be shown:

[test0-0] topic-partitions are assigned to this consumer
Current offset is 3 committed offset is ->OffsetAndMetadata{offset=3, metadata=''}
Setting it to the end 


Now we can try to read previous messages by changing the constructor parameter:

public static void main(String[] args) {
    final MessageConsumer<String, String> testConsumer = new MessageConsumer<>("test0", "testGroup", 2);
    testConsumer.startReceiving( (k,v) -> System.out.println("received:"+v) );
}

- it's now 2 so we will be reading from offset 2. Let's restart it again:

[test0-0] topic-partitions are assigned to this consumer
Current offset is 3 committed offset is ->OffsetAndMetadata{offset=3, metadata=''}
Resetting offset to 2
received:Test 3 Tue Jan 23 15:36:29 CET 2018

- now last previous message with the offset 2 was read.

6. The end

Source code can be downloaded from here.