BulkProcessor creation using RestHighLevelClient compilation error on ES5.6

[Rappahannock]

I have an issue when compiling the following code on ES5.6.6:

BulkProcessor bulkProcessor = BulkProcessor.builder(restHighLevelClient::bulkAsync, listener).build();

The following compilation error occurs:

/test-bulkprocessor/src/main/java/com/rappahannock/elasticsearch/ESConfiguration.java: incompatible types: org.elasticsearch.client.Client is not a functional interface
    multiple non-overriding abstract methods found in interface org.elasticsearch.client.Client

The same code compiles without error on ES6.2.4.

But in my project I cannot switch to a higher client version, because server version cannot be easily updated.

Test repository for error reproduction:

Rappahannock/test-bulkprocessor

[Rappahannock]

Similar problem has been discussed in the following topics:

Headache with RestHighLevelClient and bulkprocessor

BulkProcessor.Builder builder = BulkProcessor.builder(client::bulkAsync, new BulkProcessorListener()); this line is complaining The target type of this expression must be a functional interface the error is this part: builder(client::bulkAsync, any idea? please help import org.elasticsearch.client.RestHighLevelClient; RestHighLevelClient client; environment: java 8, eclipse

[BUG] You can use BulkProcessor with high-level client in 5.6 version

Hi team, Thanks in advance I try to create a BulkProcessor like https://www.elastic.co/guide/en/elasticsearch/client/java-rest/5.6/java-rest-high-document-bulk.html with the high-level client in the version 5.6 but I get this compilation error at this point: ... RestHighLevelClient rHLClient ... BulkProcessor bulkProcessor = BulkProcessor.builder(rHLClient::bulkAsync, listener,threadPool).build(); "Client is not a functional interface" If there is a better way to use that, please advis…

But I don't find there how to solve my problem.

This comment suggests that it's a bug in Eclipse IDE: [BUG] You can use BulkProcessor with high-level client in 5.6 version - #9 by juanaramis

But I compile in another IDE, also I compile from command line using maven 3.0.4.

Here is an extract from my maven compilation log:

Apache Maven 3.0.4 (r1232337; 2012-01-17 12:44:56+0400)
Maven home: F:\1_Tools\apache-maven-3.0.4
Java version: 1.8.0_77, vendor: Oracle Corporation
Java home: C:\Program Files\Java\jdk1.8.0_77\jre
Default locale: en_US, platform encoding: Cp1251
OS name: "windows 7", version: "6.1", arch: "amd64", family: "dos"
[INFO] Error stacktraces are turned on.
[DEBUG] Reading global settings from F:\1_Tools\apache-maven-3.0.4\conf\settings.xml
[DEBUG] Reading user settings from C:\...\.m2\settings.xml
[DEBUG] Using local repository at C:\m2
[DEBUG] Using manager EnhancedLocalRepositoryManager with priority 10 for C:\m2
...
[INFO] -------------------------------------------------------------
[ERROR] COMPILATION ERROR : 
[INFO] -------------------------------------------------------------
[ERROR] /F:/GIT/0_Examples/test-bulkprocessor/src/main/java/com/rappahannock/elasticsearch/ESConfiguration.java: incompatible types: org.elasticsearch.client.Client is not a functional interface
    multiple non-overriding abstract methods found in interface org.elasticsearch.client.Client
[INFO] 1 error
[INFO] -------------------------------------------------------------
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 7.067s
[INFO] Finished at: Wed Jul 18 17:28:21 GET 2018
[INFO] Final Memory: 26M/306M
[INFO] ------------------------------------------------------------------------
[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.1:compile (default-compile) on project test-bulkprocessor: Compilation failure
[ERROR] /F:/GIT/0_Examples/test-bulkprocessor/src/main/java/com/rappahannock/elasticsearch/ESConfiguration.java: incompatible types: org.elasticsearch.client.Client is not a functional interface
[ERROR] multiple non-overriding abstract methods found in interface org.elasticsearch.client.Client
[ERROR] -> [Help 1]
org.apache.maven.lifecycle.LifecycleExecutionException: Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.1:compile (default-compile) on project test-bulkprocessor: Compilation failure
/F:/GIT/0_Examples/test-bulkprocessor/src/main/java/com/rappahannock/elasticsearch/ESConfiguration.java: incompatible types: org.elasticsearch.client.Client is not a functional interface
    multiple non-overriding abstract methods found in interface org.elasticsearch.client.Client

	at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:213)
	at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:153)
	at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:145)
	at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject(LifecycleModuleBuilder.java:84)
	at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject(LifecycleModuleBuilder.java:59)
	at org.apache.maven.lifecycle.internal.LifecycleStarter.singleThreadedBuild(LifecycleStarter.java:183)
	at org.apache.maven.lifecycle.internal.LifecycleStarter.execute(LifecycleStarter.java:161)
	at org.apache.maven.DefaultMaven.doExecute(DefaultMaven.java:320)
	at org.apache.maven.DefaultMaven.execute(DefaultMaven.java:156)
	at org.apache.maven.cli.MavenCli.execute(MavenCli.java:537)
	at org.apache.maven.cli.MavenCli.doMain(MavenCli.java:196)
	at org.apache.maven.cli.MavenCli.main(MavenCli.java:141)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.codehaus.plexus.classworlds.launcher.Launcher.launchEnhanced(Launcher.java:290)
	at org.codehaus.plexus.classworlds.launcher.Launcher.launch(Launcher.java:230)
	at org.codehaus.plexus.classworlds.launcher.Launcher.mainWithExitCode(Launcher.java:409)
	at org.codehaus.plexus.classworlds.launcher.Launcher.main(Launcher.java:352)
Caused by: org.apache.maven.plugin.compiler.CompilationFailureException: Compilation failure
/F:/GIT/0_Examples/test-bulkprocessor/src/main/java/com/rappahannock/elasticsearch/ESConfiguration.java: incompatible types: org.elasticsearch.client.Client is not a functional interface
    multiple non-overriding abstract methods found in interface org.elasticsearch.client.Client

	at org.apache.maven.plugin.compiler.AbstractCompilerMojo.execute(AbstractCompilerMojo.java:858)
	at org.apache.maven.plugin.compiler.CompilerMojo.execute(CompilerMojo.java:129)
	at org.apache.maven.plugin.DefaultBuildPluginManager.executeMojo(DefaultBuildPluginManager.java:101)
	at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:209)
	... 19 more

[juanaramis]

Hi Rappahannock,

That is a eclipse bug: https://bugs.eclipse.org/bugs/show_bug.cgi?id=514884

I just write the code this way with try with resource:

try (final BulkProcessor bulkProcessor = new BulkProcessor.Builder(restHighLevelClient::bulkAsync,//<--- See BUG https://bugs.eclipse.org/bugs/show_bug.cgi?id=514884
                                                                     getBulkProcessorListener(),
                                                                     new ThreadPool(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "high-level-client").build()))
                                                            .setConcurrentRequests(props.getConcurrentRequests())
                                                            .setBulkSize(new ByteSizeValue(props.getBulkSize(), ByteSizeUnit.MB))
                                                            .setBulkActions(props.getBulkActions())
                                                            .setFlushInterval(TimeValue.timeValueSeconds(props.getFlushIntervalSeconds()))
                                                            .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(props.getConnectionRequestTimeOut() * 1000), props.getMaxNumberOfRetry()))
                                                            .build())
        {

....

Try it, and please let me know if that solved your issue.

Juan Cobas

[Rappahannock]

Hi juanaramis,

Thank you for your input.

As I mentioned in my previous comment, I don't use Eclipse, I compile in another IDE and from command line. So I don't think it can be Eclipse bug.

But your piece of code is helpful anyway, because it works!
But actually you don't need 'try with resource' for it to work.

The following piece of code works fine too:

        BulkProcessor bulkProcessor = new BulkProcessor.Builder(
                             restHighLevelClient::bulkAsync,
                             new BulkProcessorListener(),
                             new ThreadPool(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "high-level-client").build()))
                                .setBulkActions(bulkActionsCount)
                                .build();

The only difference with my original code here is that I need to supply ThreadPool by myself, which so far I didn't want to do.

[Rappahannock]

Here is a part from BulkProcessor code from ES6.2.4 which is not yet in v5.6.6

    public static BulkProcessor.Builder builder(BiConsumer<BulkRequest, ActionListener<BulkResponse>> consumer, BulkProcessor.Listener listener) {
        Objects.requireNonNull(consumer, "consumer");
        Objects.requireNonNull(listener, "listener");
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = Scheduler.initScheduler(Settings.EMPTY);
        return new BulkProcessor.Builder(consumer, listener, (delay, executor, command) -> {
            return scheduledThreadPoolExecutor.schedule(command, delay.millis(), TimeUnit.MILLISECONDS);
        }, () -> {
            Scheduler.terminate(scheduledThreadPoolExecutor, 10L, TimeUnit.SECONDS);
        }, null);
    }

So I guess this is what makes my original code work on ES6.2.4.

BulkProcessor bulkProcessor = BulkProcessor.builder(restHighLevelClient::bulkAsync, listener).build();

But I cannot create threadPool the same way, because class Scheduler doesn't exist in v5.6.6.
So I think I need to dig into code of threadPool creation in ES6.2.4 to understand how can I create it in similar way and use this code on ES5.6.6

        BulkProcessor bulkProcessor =
            new BulkProcessor.Builder(restHighLevelClient::bulkAsync, new BulkProcessorListener(), threadPool)
                .setBulkActions(bulkActionsCount)
                .build();

[Rappahannock]

I found these related issues:

Issue: BulkProcessor can't be used with empty settings in the high level client

Decouple BulkProcessor from ThreadPool

So I came to a conclusion, that is a good solution for v.5.6.6 to use the following code

BulkProcessor bulkProcessor = new BulkProcessor.Builder(
            restHighLevelClient::bulkAsync, new BulkProcessorListener(),
            new ThreadPool(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "high-level-client").build()))
            .setBulkActions(bulkActionsCount)
            .build();

@juanaramis, thanks again for the hint.

[juanaramis]

You are welcome