Addition of rat and spolessApply
* All offending files were also corrected.
diff --git a/build.gradle b/build.gradle
index 114b3cd..124ef4b 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1,9 +1,57 @@
+/*
+ * 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.
+ */
+buildscript {
+ repositories {
+ maven { url "https://plugins.gradle.org/m2/" }
+ maven { url "https://dl.bintray.com/palantir/releases" }
+ jcenter()
+ }
+ dependencies {
+ classpath 'com.diffplug.spotless:spotless-plugin-gradle:3.21.1'
+ classpath 'com.github.ben-manes:gradle-versions-plugin:0.21.0'
+ classpath 'com.netflix.nebula:gradle-lint-plugin:11.4.4'
+ classpath 'com.netflix.nebula:nebula-project-plugin:6.0.2'
+ classpath 'gradle.plugin.com.palantir.gradle.docker:gradle-docker:0.22.1'
+ classpath 'io.spring.gradle:dependency-management-plugin:1.0.7.RELEASE'
+ classpath 'org.ajoberstar.grgit:grgit-gradle:3.1.1'
+ classpath 'org.nosphere.apache:creadur-rat-gradle:0.4.0'
+ classpath 'org.sonarsource.scanner.gradle:sonarqube-gradle-plugin:2.7'
+ }
+}
plugins {
id 'java'
}
-group 'org.apache.geode'
-version '1.0-SNAPSHOT'
+apply plugin: 'wrapper'
+apply plugin: 'nebula.facet'
+apply plugin: 'java-library'
+apply plugin: 'idea'
+apply plugin: 'eclipse'
+
+tasks.register('devBuild') {
+ description "A convenience target for a typical developer workflow: apply spotless and assemble all classes."
+ dependsOn tasks.named('assemble')
+ // Each subproject injects its SpotlessApply as a dependency to this task in the standard config
+}
+
+
+apply from: "${rootDir}/${scriptDir}/spotless.gradle"
+apply from: "${scriptDir}/rat.gradle"
+
sourceCompatibility = 1.8
@@ -12,22 +60,22 @@
}
dependencies {
-
- compile 'org.apache.geode:geode-core:1.10.0'
- compile 'org.apache.geode:geode-cq:1.10.0'
+ compile('org.apache.geode:geode-core:1.10.0')
+ compile('org.apache.geode:geode-cq:1.10.0')
compile(group: 'org.apache.kafka', name: 'connect-api', version: '2.3.1')
- compile group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.13.0'
- compile group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.13.0'
+ compile(group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.13.0')
+ compile(group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.13.0')
testCompile(group: 'org.apache.kafka', name: 'kafka_2.12', version: '2.3.1')
testCompile(group: 'org.apache.kafka', name: 'kafka-streams-test-utils', version: '1.1.0')
testCompile(group: 'org.apache.curator', name: 'curator-framework', version: '4.2.0')
testCompile(group: 'org.apache.kafka', name: 'connect-runtime', version: '2.3.1')
- testCompile group: 'junit', name: 'junit', version: '4.12'
- testCompile 'org.mockito:mockito-core:3.2.4'
- testCompile 'pl.pragmatists:JUnitParams:1.1.1'
+ testCompile(group: 'junit', name: 'junit', version: '4.12')
+ testCompile('org.mockito:mockito-core:3.2.4')
+ testCompile('pl.pragmatists:JUnitParams:1.1.1')
testImplementation 'org.awaitility:awaitility:4.0.2'
}
+
diff --git a/etc/eclipse-java-google-style.xml b/etc/eclipse-java-google-style.xml
new file mode 100644
index 0000000..629ef19
--- /dev/null
+++ b/etc/eclipse-java-google-style.xml
@@ -0,0 +1,337 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<profiles version="13">
+<profile kind="CodeFormatterProfile" name="GoogleStyle" version="13">
+<setting id="org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode" value="enabled"/>
+<setting id="org.eclipse.jdt.core.compiler.codegen.targetPlatform" value="1.7"/>
+<setting id="org.eclipse.jdt.core.compiler.compliance" value="1.7"/>
+<setting id="org.eclipse.jdt.core.compiler.problem.assertIdentifier" value="error"/>
+<setting id="org.eclipse.jdt.core.compiler.problem.enumIdentifier" value="error"/>
+<setting id="org.eclipse.jdt.core.compiler.source" value="1.7"/>
+<setting id="org.eclipse.jdt.core.formatter.align_type_members_on_columns" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_field" value="1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_field.count_dependent" value="1585|-1|1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_local_variable" value="1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_local_variable.count_dependent" value="1585|-1|1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_method" value="1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_method.count_dependent" value="1585|-1|1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_package" value="1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_package.count_dependent" value="1585|-1|1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_parameter" value="1040"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_parameter.count_dependent" value="1040|-1|1040"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_type" value="1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_annotations_on_type.count_dependent" value="1585|-1|1585"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_allocation_expression" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_allocation_expression.count_dependent" value="16|5|80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_annotation" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_annotation.count_dependent" value="16|-1|16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_enum_constant" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_enum_constant.count_dependent" value="16|-1|16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_explicit_constructor_call" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_explicit_constructor_call.count_dependent" value="16|5|80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_method_invocation" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_method_invocation.count_dependent" value="16|5|80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_qualified_allocation_expression" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_arguments_in_qualified_allocation_expression.count_dependent" value="16|4|80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_assignment" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_binary_expression" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_binary_expression.count_dependent" value="16|-1|16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_cascading_method_invocation_with_arguments" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_cascading_method_invocation_with_arguments.count_dependent" value="16|-1|16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_compact_if" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_conditional_expression" value="80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_enum_constants" value="48"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_enum_constants.count_dependent" value="16|5|48"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_expressions_in_array_initializer" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_expressions_in_array_initializer.count_dependent" value="16|5|80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_field_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_for_statement" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_generic_type_arguments" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_generic_type_arguments.count_dependent" value="16|-1|16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_local_variable_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_method_declaration" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_multiple_fields" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_multiple_fields.count_dependent" value="16|-1|16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_new_anonymous_class" value="20"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_parameters_in_constructor_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_parameters_in_constructor_declaration.count_dependent" value="16|5|80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_parameters_in_method_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_parameters_in_method_declaration.count_dependent" value="16|5|80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_resources_in_try" value="80"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_selector_in_method_invocation" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_selector_in_method_invocation.count_dependent" value="16|4|48"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superclass_in_type_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superinterfaces_in_enum_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superinterfaces_in_enum_declaration.count_dependent" value="16|4|49"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superinterfaces_in_type_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_superinterfaces_in_type_declaration.count_dependent" value="16|4|48"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_throws_clause_in_constructor_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_throws_clause_in_constructor_declaration.count_dependent" value="16|4|48"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_throws_clause_in_method_declaration" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_throws_clause_in_method_declaration.count_dependent" value="16|4|48"/>
+<setting id="org.eclipse.jdt.core.formatter.alignment_for_union_type_in_multicatch" value="16"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_after_imports" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_after_package" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_field" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_first_class_body_declaration" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_imports" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_member_type" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_method" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_new_chunk" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_before_package" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_between_import_groups" value="1"/>
+<setting id="org.eclipse.jdt.core.formatter.blank_lines_between_type_declarations" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_annotation_type_declaration" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_anonymous_type_declaration" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_array_initializer" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_block" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_block_in_case" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_constructor_declaration" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_enum_constant" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_enum_declaration" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_method_declaration" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_switch" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.brace_position_for_type_declaration" value="end_of_line"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.clear_blank_lines_in_block_comment" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.clear_blank_lines_in_javadoc_comment" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_block_comments" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_html" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_javadoc_comments" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_line_comments" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.format_source_code" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.indent_parameter_description" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.indent_root_tags" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.insert_new_line_before_root_tags" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.insert_new_line_for_parameter" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.line_length" value="100"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.new_lines_at_block_boundaries" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.new_lines_at_javadoc_boundaries" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.comment.preserve_white_space_between_code_and_line_comments" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.comment_new_line_at_start_of_html_paragraph" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.compact_else_if" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.continuation_indentation" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.continuation_indentation_for_array_initializer" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.disabling_tag" value="@formatter:off"/>
+<setting id="org.eclipse.jdt.core.formatter.enabling_tag" value="@formatter:on"/>
+<setting id="org.eclipse.jdt.core.formatter.force_if_else_statement_brace" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.format_guardian_clause_on_one_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.format_line_comment_starting_on_first_column" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_annotation_declaration_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_enum_constant_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_enum_declaration_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_body_declarations_compare_to_type_header" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_breaks_compare_to_cases" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_empty_lines" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_statements_compare_to_block" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_statements_compare_to_body" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_switchstatements_compare_to_cases" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indent_switchstatements_compare_to_switch" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.indentation.size" value="4"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_field" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_local_variable" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_method" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_package" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_parameter" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_annotation_on_type" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_label" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_after_opening_brace_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_at_end_of_file_if_missing" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_catch_in_try_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_else_in_if_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_finally_in_try_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_before_while_in_do_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_annotation_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_anonymous_type_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_block" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_enum_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_method_body" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_new_line_in_empty_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_and_in_type_parameter" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_assignment_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_at_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_at_in_annotation_type_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_binary_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_angle_bracket_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_angle_bracket_in_type_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_brace_in_block" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_closing_paren_in_cast" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_assert" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_case" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_colon_in_labeled_statement" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_allocation_expression" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_annotation" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_array_initializer" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_constructor_declaration_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_constructor_declaration_throws" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_enum_constant_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_enum_declarations" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_explicitconstructorcall_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_for_increments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_for_inits" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_method_declaration_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_method_declaration_throws" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_method_invocation_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_multiple_field_declarations" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_multiple_local_declarations" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_parameterized_type_reference" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_superinterfaces" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_type_arguments" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comma_in_type_parameters" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_comment_prefix" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_ellipsis" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_angle_bracket_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_angle_bracket_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_angle_bracket_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_brace_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_bracket_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_bracket_in_array_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_cast" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_catch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_for" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_if" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_parenthesized_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_switch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_synchronized" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_try" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_opening_paren_in_while" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_postfix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_prefix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_question_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_question_in_wildcard" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_semicolon_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_semicolon_in_try_resources" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_after_unary_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_and_in_type_parameter" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_assignment_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_at_in_annotation_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_binary_operator" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_angle_bracket_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_angle_bracket_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_angle_bracket_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_brace_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_bracket_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_bracket_in_array_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_cast" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_catch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_for" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_if" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_parenthesized_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_switch" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_synchronized" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_try" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_closing_paren_in_while" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_assert" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_case" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_default" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_colon_in_labeled_statement" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_constructor_declaration_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_constructor_declaration_throws" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_enum_constant_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_enum_declarations" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_explicitconstructorcall_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_for_increments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_for_inits" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_method_declaration_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_method_declaration_throws" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_method_invocation_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_multiple_field_declarations" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_multiple_local_declarations" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_superinterfaces" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_comma_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_ellipsis" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_angle_bracket_in_parameterized_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_angle_bracket_in_type_arguments" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_angle_bracket_in_type_parameters" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_annotation_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_anonymous_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_array_initializer" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_block" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_constructor_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_enum_constant" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_enum_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_method_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_switch" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_brace_in_type_declaration" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_bracket_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_bracket_in_array_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_bracket_in_array_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_annotation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_annotation_type_member_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_catch" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_for" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_if" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_parenthesized_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_switch" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_synchronized" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_try" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_opening_paren_in_while" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_parenthesized_expression_in_return" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_parenthesized_expression_in_throw" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_postfix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_prefix_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_question_in_conditional" value="insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_question_in_wildcard" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_semicolon" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_semicolon_in_for" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_semicolon_in_try_resources" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_before_unary_operator" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_brackets_in_array_type_reference" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_braces_in_array_initializer" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_brackets_in_array_allocation_expression" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_annotation_type_member_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_constructor_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_enum_constant" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_method_declaration" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.insert_space_between_empty_parens_in_method_invocation" value="do not insert"/>
+<setting id="org.eclipse.jdt.core.formatter.join_lines_in_comments" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.join_wrapped_lines" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_else_statement_on_same_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_empty_array_initializer_on_one_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_imple_if_on_one_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.keep_then_statement_on_same_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.lineSplit" value="100"/>
+<setting id="org.eclipse.jdt.core.formatter.never_indent_block_comments_on_first_column" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.never_indent_line_comments_on_first_column" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.number_of_blank_lines_at_beginning_of_method_body" value="0"/>
+<setting id="org.eclipse.jdt.core.formatter.number_of_empty_lines_to_preserve" value="3"/>
+<setting id="org.eclipse.jdt.core.formatter.put_empty_statement_on_new_line" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.tabulation.char" value="space"/>
+<setting id="org.eclipse.jdt.core.formatter.tabulation.size" value="2"/>
+<setting id="org.eclipse.jdt.core.formatter.use_on_off_tags" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.use_tabs_only_for_leading_indentations" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_before_binary_operator" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_before_or_operator_multicatch" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_comment_inline_tags" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_non_simple_local_variable_annotation" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_non_simple_member_annotation" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_non_simple_package_annotation" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_non_simple_parameter_annotation" value="false"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_non_simple_type_annotation" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_outer_expressions_when_nested" value="true"/>
+<setting id="org.eclipse.jdt.core.formatter.wrap_prefer_two_fragments" value="false"/>
+</profile>
+</profiles>
diff --git a/etc/eclipseOrganizeImports.importorder b/etc/eclipseOrganizeImports.importorder
new file mode 100644
index 0000000..de96cae
--- /dev/null
+++ b/etc/eclipseOrganizeImports.importorder
@@ -0,0 +1,8 @@
+#Organize Import Order
+#Thu Sep 15 13:10:33 PDT 2016
+5=com.gemstone
+4=org.apache.geode
+3=
+2=javax
+1=java
+0=\#
diff --git a/etc/intellij-java-modified-google-style.xml b/etc/intellij-java-modified-google-style.xml
new file mode 100644
index 0000000..696500c
--- /dev/null
+++ b/etc/intellij-java-modified-google-style.xml
@@ -0,0 +1,255 @@
+<code_scheme name="GeodeStyle">
+ <option name="JAVA_INDENT_OPTIONS">
+ <value>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="4" />
+ <option name="TAB_SIZE" value="8" />
+ <option name="USE_TAB_CHARACTER" value="false" />
+ <option name="SMART_TABS" value="false" />
+ <option name="LABEL_INDENT_SIZE" value="0" />
+ <option name="LABEL_INDENT_ABSOLUTE" value="false" />
+ <option name="USE_RELATIVE_INDENTS" value="false" />
+ </value>
+ </option>
+ <option name="OTHER_INDENT_OPTIONS">
+ <value>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="4" />
+ <option name="TAB_SIZE" value="8" />
+ <option name="USE_TAB_CHARACTER" value="false" />
+ <option name="SMART_TABS" value="false" />
+ <option name="LABEL_INDENT_SIZE" value="0" />
+ <option name="LABEL_INDENT_ABSOLUTE" value="false" />
+ <option name="USE_RELATIVE_INDENTS" value="false" />
+ </value>
+ </option>
+ <option name="LINE_SEPARATOR" value="
" />
+ <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="99" />
+ <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="99" />
+ <option name="PACKAGES_TO_USE_IMPORT_ON_DEMAND">
+ <value />
+ </option>
+ <option name="IMPORT_LAYOUT_TABLE">
+ <value>
+ <package name="" withSubpackages="true" static="true" />
+ <emptyLine />
+ <package name="java" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="javax" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="org.apache.geode" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="com.gemstone" withSubpackages="true" static="false" />
+ </value>
+ </option>
+ <option name="STATIC_METHODS_ORDER_WEIGHT" value="5" />
+ <option name="METHODS_ORDER_WEIGHT" value="4" />
+ <option name="RIGHT_MARGIN" value="100" />
+ <option name="JD_ALIGN_PARAM_COMMENTS" value="false" />
+ <option name="JD_ALIGN_EXCEPTION_COMMENTS" value="false" />
+ <option name="JD_ADD_BLANK_AFTER_DESCRIPTION" value="false" />
+ <option name="JD_P_AT_EMPTY_LINES" value="false" />
+ <option name="JD_KEEP_EMPTY_PARAMETER" value="false" />
+ <option name="JD_KEEP_EMPTY_EXCEPTION" value="false" />
+ <option name="JD_KEEP_EMPTY_RETURN" value="false" />
+ <option name="HTML_KEEP_BLANK_LINES" value="1" />
+ <option name="HTML_ALIGN_TEXT" value="true" />
+ <option name="KEEP_LINE_BREAKS" value="false" />
+ <option name="KEEP_FIRST_COLUMN_COMMENT" value="false" />
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false" />
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1" />
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1" />
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1" />
+ <option name="BLANK_LINES_AROUND_CLASS" value="0" />
+ <option name="BLANK_LINES_AROUND_FIELD" value="1" />
+ <option name="BLANK_LINES_AFTER_CLASS_HEADER" value="1" />
+ <option name="BRACE_STYLE" value="2" />
+ <option name="CLASS_BRACE_STYLE" value="2" />
+ <option name="METHOD_BRACE_STYLE" value="2" />
+ <option name="ELSE_ON_NEW_LINE" value="true" />
+ <option name="WHILE_ON_NEW_LINE" value="true" />
+ <option name="CATCH_ON_NEW_LINE" value="true" />
+ <option name="FINALLY_ON_NEW_LINE" value="true" />
+ <option name="ALIGN_MULTILINE_PARAMETERS_IN_CALLS" value="true" />
+ <option name="ALIGN_MULTILINE_BINARY_OPERATION" value="true" />
+ <option name="ALIGN_MULTILINE_ASSIGNMENT" value="true" />
+ <option name="ALIGN_MULTILINE_TERNARY_OPERATION" value="true" />
+ <option name="ALIGN_MULTILINE_THROWS_LIST" value="true" />
+ <option name="ALIGN_MULTILINE_EXTENDS_LIST" value="true" />
+ <option name="ALIGN_MULTILINE_PARENTHESIZED_EXPRESSION" value="true" />
+ <option name="ALIGN_MULTILINE_ARRAY_INITIALIZER_EXPRESSION" value="true" />
+ <option name="SPACE_AFTER_TYPE_CAST" value="false" />
+ <option name="SPACE_BEFORE_IF_PARENTHESES" value="false" />
+ <option name="SPACE_BEFORE_WHILE_PARENTHESES" value="false" />
+ <option name="SPACE_BEFORE_FOR_PARENTHESES" value="false" />
+ <option name="SPACE_BEFORE_CATCH_PARENTHESES" value="false" />
+ <option name="SPACE_BEFORE_SWITCH_PARENTHESES" value="false" />
+ <option name="SPACE_BEFORE_SYNCHRONIZED_PARENTHESES" value="false" />
+ <option name="CALL_PARAMETERS_WRAP" value="1" />
+ <option name="CALL_PARAMETERS_LPAREN_ON_NEXT_LINE" value="true" />
+ <option name="METHOD_PARAMETERS_WRAP" value="1" />
+ <option name="EXTENDS_LIST_WRAP" value="1" />
+ <option name="THROWS_LIST_WRAP" value="1" />
+ <option name="EXTENDS_KEYWORD_WRAP" value="1" />
+ <option name="THROWS_KEYWORD_WRAP" value="1" />
+ <option name="METHOD_CALL_CHAIN_WRAP" value="1" />
+ <option name="BINARY_OPERATION_WRAP" value="1" />
+ <option name="BINARY_OPERATION_SIGN_ON_NEXT_LINE" value="true" />
+ <option name="TERNARY_OPERATION_WRAP" value="1" />
+ <option name="TERNARY_OPERATION_SIGNS_ON_NEXT_LINE" value="true" />
+ <option name="FOR_STATEMENT_WRAP" value="1" />
+ <option name="ARRAY_INITIALIZER_WRAP" value="1" />
+ <option name="ASSIGNMENT_WRAP" value="5" />
+ <option name="WRAP_COMMENTS" value="true" />
+ <option name="IF_BRACE_FORCE" value="3" />
+ <option name="DOWHILE_BRACE_FORCE" value="3" />
+ <option name="WHILE_BRACE_FORCE" value="3" />
+ <option name="FOR_BRACE_FORCE" value="3" />
+ <JavaCodeStyleSettings>
+ <option name="DO_NOT_WRAP_AFTER_SINGLE_ANNOTATION" value="true" />
+ <option name="CLASS_NAMES_IN_JAVADOC" value="3" />
+ </JavaCodeStyleSettings>
+ <MarkdownNavigatorCodeStyleSettings>
+ <option name="RIGHT_MARGIN" value="72" />
+ </MarkdownNavigatorCodeStyleSettings>
+ <XML>
+ <option name="XML_KEEP_BLANK_LINES" value="1" />
+ <option name="XML_LEGACY_SETTINGS_IMPORTED" value="true" />
+ </XML>
+ <ADDITIONAL_INDENT_OPTIONS fileType="haml">
+ <option name="INDENT_SIZE" value="2" />
+ </ADDITIONAL_INDENT_OPTIONS>
+ <ADDITIONAL_INDENT_OPTIONS fileType="java">
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="4" />
+ <option name="TAB_SIZE" value="8" />
+ </ADDITIONAL_INDENT_OPTIONS>
+ <ADDITIONAL_INDENT_OPTIONS fileType="js">
+ <option name="CONTINUATION_INDENT_SIZE" value="4" />
+ </ADDITIONAL_INDENT_OPTIONS>
+ <ADDITIONAL_INDENT_OPTIONS fileType="sass">
+ <option name="INDENT_SIZE" value="2" />
+ </ADDITIONAL_INDENT_OPTIONS>
+ <ADDITIONAL_INDENT_OPTIONS fileType="yml">
+ <option name="INDENT_SIZE" value="2" />
+ </ADDITIONAL_INDENT_OPTIONS>
+ <codeStyleSettings language="ECMA Script Level 4">
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1" />
+ <option name="ALIGN_MULTILINE_PARAMETERS_IN_CALLS" value="true" />
+ <option name="ALIGN_MULTILINE_BINARY_OPERATION" value="true" />
+ <option name="ALIGN_MULTILINE_TERNARY_OPERATION" value="true" />
+ <option name="ALIGN_MULTILINE_EXTENDS_LIST" value="true" />
+ <option name="ALIGN_MULTILINE_ARRAY_INITIALIZER_EXPRESSION" value="true" />
+ <option name="CALL_PARAMETERS_WRAP" value="1" />
+ <option name="METHOD_PARAMETERS_WRAP" value="1" />
+ <option name="EXTENDS_LIST_WRAP" value="1" />
+ <option name="EXTENDS_KEYWORD_WRAP" value="1" />
+ <option name="METHOD_CALL_CHAIN_WRAP" value="1" />
+ <option name="BINARY_OPERATION_WRAP" value="1" />
+ <option name="BINARY_OPERATION_SIGN_ON_NEXT_LINE" value="true" />
+ <option name="TERNARY_OPERATION_WRAP" value="1" />
+ <option name="TERNARY_OPERATION_SIGNS_ON_NEXT_LINE" value="true" />
+ <option name="FOR_STATEMENT_WRAP" value="1" />
+ <option name="ARRAY_INITIALIZER_WRAP" value="1" />
+ <option name="ASSIGNMENT_WRAP" value="5" />
+ <option name="WRAP_COMMENTS" value="true" />
+ <option name="IF_BRACE_FORCE" value="3" />
+ <option name="DOWHILE_BRACE_FORCE" value="3" />
+ <option name="WHILE_BRACE_FORCE" value="3" />
+ <option name="FOR_BRACE_FORCE" value="3" />
+ <option name="PARENT_SETTINGS_INSTALLED" value="true" />
+ </codeStyleSettings>
+ <codeStyleSettings language="JAVA">
+ <option name="RIGHT_MARGIN" value="100" />
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false" />
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1" />
+ <option name="CALL_PARAMETERS_WRAP" value="1" />
+ <option name="METHOD_PARAMETERS_WRAP" value="1" />
+ <option name="EXTENDS_LIST_WRAP" value="1" />
+ <option name="THROWS_LIST_WRAP" value="1" />
+ <option name="EXTENDS_KEYWORD_WRAP" value="1" />
+ <option name="THROWS_KEYWORD_WRAP" value="1" />
+ <option name="METHOD_CALL_CHAIN_WRAP" value="1" />
+ <option name="BINARY_OPERATION_WRAP" value="1" />
+ <option name="BINARY_OPERATION_SIGN_ON_NEXT_LINE" value="true" />
+ <option name="TERNARY_OPERATION_WRAP" value="1" />
+ <option name="TERNARY_OPERATION_SIGNS_ON_NEXT_LINE" value="true" />
+ <option name="FOR_STATEMENT_WRAP" value="1" />
+ <option name="ARRAY_INITIALIZER_WRAP" value="1" />
+ <option name="ASSIGNMENT_WRAP" value="5" />
+ <option name="IF_BRACE_FORCE" value="3" />
+ <option name="DOWHILE_BRACE_FORCE" value="3" />
+ <option name="WHILE_BRACE_FORCE" value="3" />
+ <option name="FOR_BRACE_FORCE" value="3" />
+ <indentOptions>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="4" />
+ <option name="TAB_SIZE" value="8" />
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="JavaScript">
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1" />
+ <option name="ALIGN_MULTILINE_PARAMETERS_IN_CALLS" value="true" />
+ <option name="ALIGN_MULTILINE_BINARY_OPERATION" value="true" />
+ <option name="ALIGN_MULTILINE_TERNARY_OPERATION" value="true" />
+ <option name="ALIGN_MULTILINE_ARRAY_INITIALIZER_EXPRESSION" value="true" />
+ <option name="CALL_PARAMETERS_WRAP" value="1" />
+ <option name="METHOD_PARAMETERS_WRAP" value="1" />
+ <option name="METHOD_CALL_CHAIN_WRAP" value="1" />
+ <option name="BINARY_OPERATION_WRAP" value="1" />
+ <option name="BINARY_OPERATION_SIGN_ON_NEXT_LINE" value="true" />
+ <option name="TERNARY_OPERATION_WRAP" value="1" />
+ <option name="TERNARY_OPERATION_SIGNS_ON_NEXT_LINE" value="true" />
+ <option name="FOR_STATEMENT_WRAP" value="1" />
+ <option name="ARRAY_INITIALIZER_WRAP" value="1" />
+ <option name="ASSIGNMENT_WRAP" value="5" />
+ <option name="WRAP_COMMENTS" value="true" />
+ <option name="IF_BRACE_FORCE" value="3" />
+ <option name="DOWHILE_BRACE_FORCE" value="3" />
+ <option name="WHILE_BRACE_FORCE" value="3" />
+ <option name="FOR_BRACE_FORCE" value="3" />
+ <option name="PARENT_SETTINGS_INSTALLED" value="true" />
+ </codeStyleSettings>
+ <codeStyleSettings language="PHP">
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false" />
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1" />
+ <option name="BLANK_LINES_AFTER_CLASS_HEADER" value="1" />
+ <option name="ALIGN_MULTILINE_ASSIGNMENT" value="true" />
+ <option name="ALIGN_MULTILINE_TERNARY_OPERATION" value="true" />
+ <option name="ALIGN_MULTILINE_THROWS_LIST" value="true" />
+ <option name="ALIGN_MULTILINE_EXTENDS_LIST" value="true" />
+ <option name="ALIGN_MULTILINE_PARENTHESIZED_EXPRESSION" value="true" />
+ <option name="CALL_PARAMETERS_WRAP" value="1" />
+ <option name="METHOD_PARAMETERS_WRAP" value="1" />
+ <option name="EXTENDS_LIST_WRAP" value="1" />
+ <option name="THROWS_LIST_WRAP" value="1" />
+ <option name="EXTENDS_KEYWORD_WRAP" value="1" />
+ <option name="THROWS_KEYWORD_WRAP" value="1" />
+ <option name="METHOD_CALL_CHAIN_WRAP" value="1" />
+ <option name="BINARY_OPERATION_WRAP" value="1" />
+ <option name="BINARY_OPERATION_SIGN_ON_NEXT_LINE" value="true" />
+ <option name="TERNARY_OPERATION_WRAP" value="1" />
+ <option name="TERNARY_OPERATION_SIGNS_ON_NEXT_LINE" value="true" />
+ <option name="FOR_STATEMENT_WRAP" value="1" />
+ <option name="ARRAY_INITIALIZER_WRAP" value="1" />
+ <option name="ASSIGNMENT_WRAP" value="5" />
+ <option name="WRAP_COMMENTS" value="true" />
+ <option name="IF_BRACE_FORCE" value="3" />
+ <option name="DOWHILE_BRACE_FORCE" value="3" />
+ <option name="WHILE_BRACE_FORCE" value="3" />
+ <option name="FOR_BRACE_FORCE" value="3" />
+ <option name="PARENT_SETTINGS_INSTALLED" value="true" />
+ <indentOptions>
+ <option name="INDENT_SIZE" value="4" />
+ <option name="CONTINUATION_INDENT_SIZE" value="8" />
+ <option name="TAB_SIZE" value="4" />
+ <option name="USE_TAB_CHARACTER" value="false" />
+ <option name="SMART_TABS" value="false" />
+ <option name="LABEL_INDENT_SIZE" value="0" />
+ <option name="LABEL_INDENT_ABSOLUTE" value="false" />
+ <option name="USE_RELATIVE_INDENTS" value="false" />
+ </indentOptions>
+ </codeStyleSettings>
+</code_scheme>
diff --git a/gradle.properties b/gradle.properties
new file mode 100644
index 0000000..7e073d9
--- /dev/null
+++ b/gradle.properties
@@ -0,0 +1,26 @@
+# 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.
+
+version = '1.0-SNAPSHOT'
+group = 'org.apache.geode'
+
+
+# 'apply from:' location for gradle scripts, relative to the project root. Specified here so that
+# it may be overridden by external projects or custom develop environment configurations
+scriptDir = gradle
+
+productName = Apache Geode Kafka Connector
+productOrg = Apache Software Foundation (ASF)
+
diff --git a/gradle/rat.gradle b/gradle/rat.gradle
new file mode 100644
index 0000000..75ccac3
--- /dev/null
+++ b/gradle/rat.gradle
@@ -0,0 +1,93 @@
+/*
+ * 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.
+ */
+
+apply plugin: "org.nosphere.apache.rat"
+
+rat {
+ inputDir = rootDir
+ excludes = [
+ // git
+ '.git/**',
+ '**/.gitignore',
+ '**/.gitkeep',
+
+ // gradle
+ '**/.gradle/**',
+ 'gradlew',
+ 'gradlew.bat',
+ 'gradle/wrapper/gradle-wrapper.properties',
+ 'caches/**',
+ 'daemon/**',
+ 'native/**',
+ 'wrapper/**',
+ '**/build/**',
+ '**/build-*/**',
+ '.buildinfo',
+ // IDE
+ 'etc/eclipse-java-google-style.xml',
+ 'etc/intellij-java-modified-google-style.xml',
+ 'etc/eclipseOrganizeImports.importorder',
+ '**/.project',
+ '**/.classpath',
+ '**/.settings/**',
+ '**/build-eclipse/**',
+ '**/*.iml',
+ '**/*.ipr',
+ '**/*.iws',
+ '.idea/**',
+ '**/tags',
+ '**/out/**',
+
+ // text files
+ '**/*.fig',
+ '**/*.txt',
+ '**/*.md',
+ '**/*.json',
+ '**/*.tx0',
+ '**/*.txo',
+ '**/*.log',
+ '**/*.patch',
+ '**/*.diff',
+ '**/*.rej',
+ '**/*.orig',
+ '**/*.MF',
+
+ // binary files
+ '**/*.cer',
+ '**/*.dia',
+ '**/*.gfs',
+ '**/*.gif',
+ '**/*.ico',
+ '**/*.jpg',
+ '**/*.keystore',
+ '**/*.pdf',
+ '**/*.png',
+ '**/*.ser',
+ '**/*.svg',
+ '**/*.truststore',
+ '**/*.xls',
+ '**/publickeyfile',
+ '**/*.dat',
+
+ // other text files
+ '**/log4j*.xml',
+ '.java-version', // file created by `jenv local`
+ '**/META-INF/**',
+ ]
+}
+
+check.dependsOn rat
\ No newline at end of file
diff --git a/gradle/spotless.gradle b/gradle/spotless.gradle
new file mode 100644
index 0000000..70d3c95
--- /dev/null
+++ b/gradle/spotless.gradle
@@ -0,0 +1,163 @@
+/*
+ * 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.
+ */
+
+
+// When a custom step changes, we need to bump the value passed to the method
+// bumpThisNumberIfACustomStepChanges
+// This has been historically easy to forget, however, and can cause failures in some rare cases.
+// To safeguard against this, we instead use the (partial) md5 of this file as that method input.
+def thisFile = file("${rootDir}/${scriptDir}/spotless.gradle")
+def thisFileMd5 = thisFile.text.md5() as String
+def thisFileMd5Piece = thisFileMd5.substring(0, 8)
+def thisFileIntegerHash = Integer.parseUnsignedInt(thisFileMd5Piece, 16)
+logger.debug("Using partial md5 (${thisFileIntegerHash}) of file ${thisFile} as spotless bump.")
+
+project.ext.set("spotless-file-hash", thisFileIntegerHash)
+
+
+apply plugin: "com.diffplug.gradle.spotless"
+spotless {
+ lineEndings = 'unix'
+ java {
+ target project.fileTree(project.projectDir) {
+ include '**/*.java'
+ exclude '**/generated-src/**'
+ exclude '**/build/**'
+ }
+
+ // As the method name suggests, bump this number if any of the below "custom" rules change.
+ // Spotless will not run on unchanged files unless this number changes.
+ bumpThisNumberIfACustomStepChanges(project.ext.'spotless-file-hash')
+
+ removeUnusedImports()
+
+ custom 'Remove commented-out import statements', {
+ it.replaceAll(/\n\/\/ import .*?;.*/, '')
+ }
+ custom 'Refuse wildcard imports', {
+ // Wildcard imports can't be resolved by spotless itself.
+ // This will require the developer themselves to adhere to best practices.
+ if (it =~ /\nimport .*\*;/) {
+ throw new AssertionError("Do not use wildcard imports. 'spotlessApply' cannot resolve this issue.")
+ }
+ }
+// custom 'Refuse Awaitility import', {
+// if (it =~ /import\s+(static\s+)?org.awaitility.Awaitility.*/) {
+// throw new AssertionError("Do not use Awaitility.await(). Use GeodeAwaitility.await() instead. 'spotlessApply' cannot resolve this issue.")
+// }
+// }
+ importOrderFile "${rootDir}/${scriptDir}/../etc/eclipseOrganizeImports.importorder"
+
+ custom 'Remove unhelpful javadoc stubs', {
+ // e.g., remove the following lines:
+ // "* @param paramName"
+ // "* @throws ExceptionType"
+ // "* @return returnType"'
+ // Multiline to allow anchors on newlines
+ it.replaceAll(/(?m)^ *\* *@(?:param|throws|return) *\w* *\n/, '')
+ }
+ custom 'Remove any empty Javadocs and block comments', {
+ // Matches any /** [...] */ or /* [...] */ that contains:
+ // (a) only whitespace
+ // (b) trivial information, such as "@param paramName" or @throws ExceptionType
+ // without any additional information. This information is implicit in the signature.
+ it.replaceAll(/\/\*+\s*\n(\s*\*\s*\n)*\s*\*+\/\s*\n/, '')
+ }
+
+ // Enforce style modifier order
+ custom 'Modifier ordering', {
+ def modifierRanking = [
+ "public" : 1,
+ "protected" : 2,
+ "private" : 3,
+ "abstract" : 4,
+ "default" : 5,
+ "static" : 6,
+ "final" : 7,
+ "transient" : 8,
+ "volatile" : 9,
+ "synchronized": 10,
+ "native" : 11,
+ "strictfp" : 12]
+ // Find any instance of multiple modifiers. Lead with a non-word character to avoid
+ // accidental matching against for instance, "an alternative default value"
+ it.replaceAll(/\W(?:public |protected |private |abstract |default |static |final |transient |volatile |synchronized |native |strictfp ){2,}/, {
+ // Do not replace the leading non-word character. Identify the modifiers
+ it.replaceAll(/(?:public |protected |private |abstract |default |static |final |transient |volatile |synchronized |native |strictfp ){2,}/, {
+ // Sort the modifiers according to the ranking above
+ it.split().sort({ modifierRanking[it] }).join(' ') + ' '
+ }
+ )
+ }
+ )
+ }
+
+
+ // Notes on eclipse formatter version:
+ // 4.6.3 is consistent with existing / previous behavior.
+ // 4.7.1 works, but had different default whitespace rules, notably with mid-ternary linebreak.
+ // 4.7.2 exists but is currently incompatible with our style file, raising NPEs.
+
+ // The format file is relative to geode-core and not the root project as the root project would change
+ // if Geode and submodules are included as part of a different gradle project.
+ eclipse('4.6.3').configFile "${rootDir}/${scriptDir}/../etc/eclipse-java-google-style.xml"
+ trimTrailingWhitespace()
+ endWithNewline()
+ }
+
+
+ groovyGradle {
+ target project.fileTree(project.projectDir) {
+ include '**/*.gradle'
+ exclude '**/generated-src/**'
+ exclude '**/build/**'
+ }
+
+ // As the method name suggests, bump this number if any of the below "custom" rules change.
+ // Spotless will not run on unchanged files unless this number changes.
+ bumpThisNumberIfACustomStepChanges(project.ext.'spotless-file-hash')
+
+ custom 'Use single-quote in project directives.', {
+ it.replaceAll(/project\(":([^"]*)"\)/, 'project(\':$1\')')
+ }
+
+ custom 'Use parenthesis in single-line gradle dependency declarations.', {
+ it.replaceAll(/\n(\s*\S*(?:[cC]ompile|[rR]untime)(?:Only)?) (?!\()([^{\n]*)\n/, { original, declaration, dep ->
+ "\n${declaration}(${dep})\n"
+ })
+ }
+
+ custom 'Do not pad spaces before parenthesis in gradle dependency declaration.', {
+ it.replaceAll(/\n(\s*\S*(?:[cC]ompile|[rR]untime)(?:Only)?) +\(/, '\n$1(')
+ }
+
+ indentWithSpaces(2)
+ }
+}
+
+// If we add more languages to Spotless, add them to 'compileXYZ' trigger below
+afterEvaluate {
+ // Not all projects are java projects. findByName could return null, so use the null-safe ?. operator
+ project.tasks.findByName('compileJava')?.mustRunAfter(spotlessCheck)
+ project.tasks.findByName('compileJava')?.mustRunAfter(spotlessApply)
+
+ // Within the configure block, 'project' refers to the task-owning project, in this case rootProject
+ def thisProjectScoped = project
+ rootProject.tasks.named('devBuild').configure {
+ dependsOn thisProjectScoped.tasks.named('spotlessApply')
+ }
+}
diff --git a/settings.gradle b/settings.gradle
index 48eed3b..546b8ec 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -1 +1,15 @@
+//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.
rootProject.name = 'geode-kafka-connector'
\ No newline at end of file
diff --git a/src/main/java/geode/kafka/GeodeConnectorConfig.java b/src/main/java/geode/kafka/GeodeConnectorConfig.java
index efffdcd..7702765 100644
--- a/src/main/java/geode/kafka/GeodeConnectorConfig.java
+++ b/src/main/java/geode/kafka/GeodeConnectorConfig.java
@@ -23,105 +23,106 @@
public class GeodeConnectorConfig {
- //GeodeKafka Specific Configuration
- /**
- * Identifier for each task
- */
- public static final String TASK_ID = "GEODE_TASK_ID"; //One config per task
- /**
- * Specifies which Locators to connect to Apache Geode
- */
- public static final String LOCATORS = "locators";
- public static final String DEFAULT_LOCATOR = "localhost[10334]";
- public static final String SECURITY_CLIENT_AUTH_INIT = "security-client-auth-init";
+ // GeodeKafka Specific Configuration
+ /**
+ * Identifier for each task
+ */
+ public static final String TASK_ID = "GEODE_TASK_ID"; // One config per task
+ /**
+ * Specifies which Locators to connect to Apache Geode
+ */
+ public static final String LOCATORS = "locators";
+ public static final String DEFAULT_LOCATOR = "localhost[10334]";
+ public static final String SECURITY_CLIENT_AUTH_INIT = "security-client-auth-init";
- protected final int taskId;
- protected List<LocatorHostPort> locatorHostPorts;
- private String securityClientAuthInit;
+ protected final int taskId;
+ protected List<LocatorHostPort> locatorHostPorts;
+ private String securityClientAuthInit;
- protected GeodeConnectorConfig() {
- taskId = 0;
+ protected GeodeConnectorConfig() {
+ taskId = 0;
+ }
+
+ public GeodeConnectorConfig(Map<String, String> connectorProperties) {
+ taskId = Integer.parseInt(connectorProperties.get(TASK_ID));
+ locatorHostPorts = parseLocators(connectorProperties.get(GeodeConnectorConfig.LOCATORS));
+ securityClientAuthInit = connectorProperties.get(SECURITY_CLIENT_AUTH_INIT);
+ }
+
+
+ public static Map<String, List<String>> parseTopicToRegions(String combinedBindings) {
+ // It's the same formatting, so parsing is the same going topic to region or region to topic
+ return parseRegionToTopics(combinedBindings);
+ }
+
+ /**
+ * Given a string of the form [region:topic,...] will produce a map where the key is the
+ * regionName and the value is a list of topicNames to push values to
+ *
+ * @param combinedBindings a string with similar form to "[region:topic,...], [region2:topic2,...]
+ * @return mapping of regionName to list of topics to update
+ */
+ public static Map<String, List<String>> parseRegionToTopics(String combinedBindings) {
+ if (combinedBindings == null || combinedBindings.equals("")) {
+ return new HashMap();
}
+ List<String> bindings = parseBindings(combinedBindings);
+ return bindings.stream().map(binding -> {
+ String[] regionToTopicsArray = parseBinding(binding);
+ return regionToTopicsArray;
+ }).collect(Collectors.toMap(regionToTopicsArray -> regionToTopicsArray[0],
+ regionToTopicsArray -> parseStringByComma(regionToTopicsArray[1])));
+ }
- public GeodeConnectorConfig(Map<String, String> connectorProperties) {
- taskId = Integer.parseInt(connectorProperties.get(TASK_ID));
- locatorHostPorts = parseLocators(connectorProperties.get(GeodeConnectorConfig.LOCATORS));
- securityClientAuthInit = connectorProperties.get(SECURITY_CLIENT_AUTH_INIT);
- }
+ public static List<String> parseBindings(String bindings) {
+ return Arrays.stream(bindings.split("](\\s)*,")).map((s) -> {
+ s = s.replaceAll("\\[", "");
+ s = s.replaceAll("\\]", "");
+ s = s.trim();
+ return s;
+ }).collect(Collectors.toList());
+ }
+ private static String[] parseBinding(String binding) {
+ return binding.split(":");
+ }
- public static Map<String, List<String>> parseTopicToRegions(String combinedBindings) {
- //It's the same formatting, so parsing is the same going topic to region or region to topic
- return parseRegionToTopics(combinedBindings);
- }
+ // Used to parse a string of topics or regions
+ public static List<String> parseStringByComma(String string) {
+ return parseStringBy(string, ",");
+ }
- /**
- * Given a string of the form [region:topic,...] will produce a map where the key is the
- * regionName and the value is a list of topicNames to push values to
- *
- * @param combinedBindings a string with similar form to "[region:topic,...], [region2:topic2,...]
- * @return mapping of regionName to list of topics to update
- */
- public static Map<String, List<String>> parseRegionToTopics(String combinedBindings) {
- if (combinedBindings == null || combinedBindings.equals("")){
- return new HashMap();
- }
- List<String> bindings = parseBindings(combinedBindings);
- return bindings.stream().map(binding -> {
- String[] regionToTopicsArray = parseBinding(binding);
- return regionToTopicsArray;
- }).collect(Collectors.toMap(regionToTopicsArray -> regionToTopicsArray[0], regionToTopicsArray -> parseStringByComma(regionToTopicsArray[1])));
- }
+ public static List<String> parseStringBy(String string, String regex) {
+ return Arrays.stream(string.split(regex)).map((s) -> s.trim()).collect(Collectors.toList());
+ }
- public static List<String> parseBindings(String bindings) {
- return Arrays.stream(bindings.split("](\\s)*,")).map((s) -> {
- s = s.replaceAll("\\[", "");
- s = s.replaceAll("\\]", "");
- s = s.trim();
- return s;
- }).collect(Collectors.toList());
- }
+ public static String reconstructString(Collection<String> strings) {
+ return strings.stream().collect(Collectors.joining("],["));
+ }
- private static String[] parseBinding(String binding) {
- return binding.split(":");
- }
+ List<LocatorHostPort> parseLocators(String locators) {
+ return Arrays.stream(locators.split(",")).map((s) -> {
+ String locatorString = s.trim();
+ return parseLocator(locatorString);
+ }).collect(Collectors.toList());
+ }
- //Used to parse a string of topics or regions
- public static List<String> parseStringByComma(String string) {
- return parseStringBy(string, ",");
- }
+ private LocatorHostPort parseLocator(String locatorString) {
+ String[] splits = locatorString.split("\\[");
+ String locator = splits[0];
+ int port = Integer.parseInt(splits[1].replace("]", ""));
+ return new LocatorHostPort(locator, port);
+ }
- public static List<String> parseStringBy(String string, String regex) {
- return Arrays.stream(string.split(regex)).map((s) -> s.trim()).collect(Collectors.toList());
- }
+ public int getTaskId() {
+ return taskId;
+ }
- public static String reconstructString(Collection<String> strings) {
- return strings.stream().collect(Collectors.joining("],["));
- }
+ public List<LocatorHostPort> getLocatorHostPorts() {
+ return locatorHostPorts;
+ }
- List<LocatorHostPort> parseLocators(String locators) {
- return Arrays.stream(locators.split(",")).map((s) -> {
- String locatorString = s.trim();
- return parseLocator(locatorString);
- }).collect(Collectors.toList());
- }
-
- private LocatorHostPort parseLocator(String locatorString) {
- String[] splits = locatorString.split("\\[");
- String locator = splits[0];
- int port = Integer.parseInt(splits[1].replace("]", ""));
- return new LocatorHostPort(locator, port);
- }
-
- public int getTaskId() {
- return taskId;
- }
-
- public List<LocatorHostPort> getLocatorHostPorts() {
- return locatorHostPorts;
- }
-
- public String getSecurityClientAuthInit() {
- return securityClientAuthInit;
- }
+ public String getSecurityClientAuthInit() {
+ return securityClientAuthInit;
+ }
}
diff --git a/src/main/java/geode/kafka/GeodeContext.java b/src/main/java/geode/kafka/GeodeContext.java
index ff8a8c3..ba5f9e5 100644
--- a/src/main/java/geode/kafka/GeodeContext.java
+++ b/src/main/java/geode/kafka/GeodeContext.java
@@ -14,6 +14,13 @@
*/
package geode.kafka;
+import static geode.kafka.GeodeConnectorConfig.SECURITY_CLIENT_AUTH_INIT;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.kafka.connect.errors.ConnectException;
+
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.query.CqAttributes;
@@ -21,78 +28,70 @@
import org.apache.geode.cache.query.CqExistsException;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.RegionNotFoundException;
-import org.apache.kafka.connect.errors.ConnectException;
-
-import java.util.Collection;
-import java.util.List;
-
-import static geode.kafka.GeodeConnectorConfig.SECURITY_CLIENT_AUTH_INIT;
public class GeodeContext {
- private ClientCache clientCache;
+ private ClientCache clientCache;
- public GeodeContext() {
+ public GeodeContext() {}
+
+ public ClientCache connectClient(List<LocatorHostPort> locatorHostPortList,
+ String durableClientId, String durableClientTimeout, String securityAuthInit) {
+ clientCache = createClientCache(locatorHostPortList, durableClientId, durableClientTimeout,
+ securityAuthInit);
+ return clientCache;
+ }
+
+ public ClientCache connectClient(List<LocatorHostPort> locatorHostPortList,
+ String securityAuthInit) {
+ clientCache = createClientCache(locatorHostPortList, "", "", securityAuthInit);
+ return clientCache;
+ }
+
+ public ClientCache getClientCache() {
+ return clientCache;
+ }
+
+ public ClientCache createClientCache(List<LocatorHostPort> locators, String durableClientName,
+ String durableClientTimeOut, String securityAuthInit) {
+ ClientCacheFactory ccf = new ClientCacheFactory();
+
+ if (securityAuthInit != null) {
+ ccf.set(SECURITY_CLIENT_AUTH_INIT, securityAuthInit);
}
-
- public ClientCache connectClient(List<LocatorHostPort> locatorHostPortList, String durableClientId, String durableClientTimeout, String securityAuthInit) {
- clientCache = createClientCache(locatorHostPortList, durableClientId, durableClientTimeout, securityAuthInit);
- return clientCache;
+ if (!durableClientName.equals("")) {
+ ccf.set("durable-client-id", durableClientName)
+ .set("durable-client-timeout", durableClientTimeOut);
}
+ // currently we only allow using the default pool.
+ // If we ever want to allow adding multiple pools we'll have to configure pool factories
+ ccf.setPoolSubscriptionEnabled(true);
- public ClientCache connectClient(List<LocatorHostPort> locatorHostPortList, String securityAuthInit) {
- clientCache = createClientCache(locatorHostPortList, "", "", securityAuthInit);
- return clientCache;
+ for (LocatorHostPort locator : locators) {
+ ccf.addPoolLocator(locator.getHostName(), locator.getPort());
}
+ return ccf.create();
+ }
- public ClientCache getClientCache() {
- return clientCache;
+ public CqQuery newCq(String name, String query, CqAttributes cqAttributes, boolean isDurable)
+ throws ConnectException {
+ try {
+ CqQuery cq = clientCache.getQueryService().newCq(name, query, cqAttributes, isDurable);
+ cq.execute();
+ return cq;
+ } catch (RegionNotFoundException | CqException | CqExistsException e) {
+ throw new ConnectException(e);
}
+ }
- /**
- *
- * @param locators
- * @param durableClientName
- * @param durableClientTimeOut
- * @return
- */
- public ClientCache createClientCache(List<LocatorHostPort> locators, String durableClientName, String durableClientTimeOut, String securityAuthInit) {
- ClientCacheFactory ccf = new ClientCacheFactory();
-
- if (securityAuthInit != null) {
- ccf.set(SECURITY_CLIENT_AUTH_INIT, securityAuthInit);
- }
- if (!durableClientName.equals("")) {
- ccf.set("durable-client-id", durableClientName)
- .set("durable-client-timeout", durableClientTimeOut);
- }
- //currently we only allow using the default pool.
- //If we ever want to allow adding multiple pools we'll have to configure pool factories
- ccf.setPoolSubscriptionEnabled(true);
-
- for (LocatorHostPort locator: locators) {
- ccf.addPoolLocator(locator.getHostName(), locator.getPort());
- }
- return ccf.create();
+ public Collection newCqWithInitialResults(String name, String query, CqAttributes cqAttributes,
+ boolean isDurable) throws ConnectException {
+ try {
+ CqQuery cq = clientCache.getQueryService().newCq(name, query, cqAttributes, isDurable);
+ return cq.executeWithInitialResults();
+ } catch (RegionNotFoundException | CqException | CqExistsException e) {
+ throw new ConnectException(e);
}
-
- public CqQuery newCq(String name, String query, CqAttributes cqAttributes, boolean isDurable) throws ConnectException {
- try {
- CqQuery cq = clientCache.getQueryService().newCq(name, query, cqAttributes, isDurable);
- cq.execute();
- return cq;
- } catch (RegionNotFoundException | CqException | CqExistsException e) {
- throw new ConnectException(e);
- }
- }
-
- public Collection newCqWithInitialResults(String name, String query, CqAttributes cqAttributes, boolean isDurable) throws ConnectException {
- try {
- CqQuery cq = clientCache.getQueryService().newCq(name, query, cqAttributes, isDurable);
- return cq.executeWithInitialResults();
- } catch (RegionNotFoundException | CqException | CqExistsException e) {
- throw new ConnectException(e);
- }
- }
+ }
}
diff --git a/src/main/java/geode/kafka/LocatorHostPort.java b/src/main/java/geode/kafka/LocatorHostPort.java
index 21d0603..191d42d 100644
--- a/src/main/java/geode/kafka/LocatorHostPort.java
+++ b/src/main/java/geode/kafka/LocatorHostPort.java
@@ -16,22 +16,23 @@
public class LocatorHostPort {
- private String hostName;
- private int port;
+ private String hostName;
+ private int port;
- public LocatorHostPort(String hostName, int port) {
- this.hostName = hostName;
- this.port = port;
- }
+ public LocatorHostPort(String hostName, int port) {
+ this.hostName = hostName;
+ this.port = port;
+ }
- public String getHostName() {
- return hostName;
- }
+ public String getHostName() {
+ return hostName;
+ }
- public int getPort() {
- return port;
- }
- public String toString() {
- return hostName + "[" + port + "]";
- }
+ public int getPort() {
+ return port;
+ }
+
+ public String toString() {
+ return hostName + "[" + port + "]";
+ }
}
diff --git a/src/main/java/geode/kafka/sink/BatchRecords.java b/src/main/java/geode/kafka/sink/BatchRecords.java
index c10faaf..0221cbe 100644
--- a/src/main/java/geode/kafka/sink/BatchRecords.java
+++ b/src/main/java/geode/kafka/sink/BatchRecords.java
@@ -14,69 +14,67 @@
*/
package geode.kafka.sink;
-import geode.kafka.source.GeodeKafkaSourceTask;
-import org.apache.geode.cache.Region;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
import org.apache.kafka.connect.sink.SinkRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import org.apache.geode.cache.Region;
/**
* A collection of records to put/remove from a region
*/
public class BatchRecords {
- private static final Logger logger = LoggerFactory.getLogger(BatchRecords.class);
+ private static final Logger logger = LoggerFactory.getLogger(BatchRecords.class);
- private Map updateMap;
- private Collection removeList;
+ private Map updateMap;
+ private Collection removeList;
- public BatchRecords() {
- this(new HashMap(), new ArrayList());
+ public BatchRecords() {
+ this(new HashMap(), new ArrayList());
+ }
+
+ /** Used for tests **/
+ public BatchRecords(Map updateMap, Collection removeList) {
+ this.updateMap = updateMap;
+ this.removeList = removeList;
+ }
+
+ public void addRemoveOperation(SinkRecord record) {
+ // if a previous operation added to the update map
+ // let's just remove it so we don't do a put and then a remove
+ // depending on the order of operations (putAll then removeAll or removeAll or putAll)...
+ // ...we could remove one of the if statements.
+ if (updateMap.containsKey(record.key())) {
+ updateMap.remove(record.key());
+ } else {
+ removeList.add(record.key());
}
+ }
- /** Used for tests**/
- public BatchRecords(Map updateMap, Collection removeList) {
- this.updateMap = updateMap;
- this.removeList = removeList;
+ public void addUpdateOperation(SinkRecord record, boolean nullValuesMeansRemove) {
+ // it's assumed the records in are order
+ // if so if a previous value was in the remove list
+ // let's not remove it at the end of this operation
+ if (nullValuesMeansRemove) {
+ if (removeList.contains(record.key())) {
+ removeList.remove(record.key());
+ }
}
+ updateMap.put(record.key(), record.value());
+ }
- public void addRemoveOperation(SinkRecord record) {
- //if a previous operation added to the update map
- //let's just remove it so we don't do a put and then a remove
- //depending on the order of operations (putAll then removeAll or removeAll or putAll)...
- //...we could remove one of the if statements.
- if (updateMap.containsKey(record.key())) {
- updateMap.remove(record.key());
- } else {
- removeList.add(record.key());
- }
+
+ public void executeOperations(Region region) {
+ if (region != null) {
+ region.putAll(updateMap);
+ region.removeAll(removeList);
+ } else {
+ logger.info("Unable to locate proxy region: " + region);
}
-
- public void addUpdateOperation(SinkRecord record, boolean nullValuesMeansRemove) {
- //it's assumed the records in are order
- //if so if a previous value was in the remove list
- // let's not remove it at the end of this operation
- if (nullValuesMeansRemove) {
- if (removeList.contains(record.key())) {
- removeList.remove(record.key());
- }
- }
- updateMap.put(record.key(), record.value());
- }
-
-
- public void executeOperations(Region region) {
- if (region != null) {
- region.putAll(updateMap);
- region.removeAll(removeList);
- }
- else {
- logger.info("Unable to locate proxy region: " + region);
- }
- }
+ }
}
diff --git a/src/main/java/geode/kafka/sink/GeodeKafkaSink.java b/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
index 1ccb385..550e8a9 100644
--- a/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
+++ b/src/main/java/geode/kafka/sink/GeodeKafkaSink.java
@@ -14,72 +14,71 @@
*/
package geode.kafka.sink;
-import geode.kafka.GeodeConnectorConfig;
-import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.sink.SinkConnector;
-import org.apache.kafka.connect.util.ConnectorUtils;
+import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
+import static geode.kafka.GeodeConnectorConfig.LOCATORS;
+import static geode.kafka.GeodeSinkConnectorConfig.DEFAULT_NULL_VALUES_MEAN_REMOVE;
+import static geode.kafka.GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
-import static geode.kafka.GeodeConnectorConfig.LOCATORS;
-import static geode.kafka.GeodeSinkConnectorConfig.DEFAULT_NULL_VALUES_MEAN_REMOVE;
-import static geode.kafka.GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE;
-import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import geode.kafka.GeodeConnectorConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.sink.SinkConnector;
-public class GeodeKafkaSink extends SinkConnector {
- private static final ConfigDef CONFIG_DEF = new ConfigDef();
- private Map<String, String> sharedProps;
+public class GeodeKafkaSink extends SinkConnector {
+ private static final ConfigDef CONFIG_DEF = new ConfigDef();
+ private Map<String, String> sharedProps;
- @Override
- public void start(Map<String, String> props) {
- sharedProps = computeMissingConfigurations(props);
+ @Override
+ public void start(Map<String, String> props) {
+ sharedProps = computeMissingConfigurations(props);
+ }
+
+ @Override
+ public Class<? extends Task> taskClass() {
+ return GeodeKafkaSinkTask.class;
+ }
+
+ @Override
+ public List<Map<String, String>> taskConfigs(int maxTasks) {
+ List<Map<String, String>> taskConfigs = new ArrayList<>();
+
+ // All tasks will build up the topic to regions map. A few might not use certain keys but we
+ // have no control over partitioning in kafka and which tasks will fire
+ for (int i = 0; i < maxTasks; i++) {
+ Map<String, String> taskProps = new HashMap<>();
+ taskProps.putAll(sharedProps);
+ taskProps.put(GeodeConnectorConfig.TASK_ID, "" + i);
+ taskConfigs.add(taskProps);
}
- @Override
- public Class<? extends Task> taskClass() {
- return GeodeKafkaSinkTask.class;
- }
+ return taskConfigs;
+ }
- @Override
- public List<Map<String, String>> taskConfigs(int maxTasks) {
- List<Map<String, String>> taskConfigs = new ArrayList<>();
+ @Override
+ public void stop() {
- //All tasks will build up the topic to regions map. A few might not use certain keys but we have no control over partitioning in kafka and which tasks will fire
- for (int i = 0; i < maxTasks; i++) {
- Map<String, String> taskProps = new HashMap<>();
- taskProps.putAll(sharedProps);
- taskProps.put(GeodeConnectorConfig.TASK_ID, "" + i);
- taskConfigs.add(taskProps);
- }
+ }
- return taskConfigs;
- }
+ @Override
+ public ConfigDef config() {
+ return CONFIG_DEF;
+ }
- @Override
- public void stop() {
-
- }
-
- @Override
- public ConfigDef config() {
- return CONFIG_DEF;
- }
-
- @Override
- public String version() {
- //TODO
- return "unknown";
- }
+ @Override
+ public String version() {
+ // TODO
+ return "unknown";
+ }
- private Map<String, String> computeMissingConfigurations(Map<String, String> props) {
- props.computeIfAbsent(LOCATORS, (key)-> DEFAULT_LOCATOR);
- props.computeIfAbsent(NULL_VALUES_MEAN_REMOVE, (key) -> DEFAULT_NULL_VALUES_MEAN_REMOVE);
- return props;
- }
+ private Map<String, String> computeMissingConfigurations(Map<String, String> props) {
+ props.computeIfAbsent(LOCATORS, (key) -> DEFAULT_LOCATOR);
+ props.computeIfAbsent(NULL_VALUES_MEAN_REMOVE, (key) -> DEFAULT_NULL_VALUES_MEAN_REMOVE);
+ return props;
+ }
}
diff --git a/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java b/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
index bcc5885..3552b2d 100644
--- a/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
+++ b/src/main/java/geode/kafka/sink/GeodeKafkaSinkTask.java
@@ -14,22 +14,23 @@
*/
package geode.kafka.sink;
-import geode.kafka.GeodeContext;
-import geode.kafka.GeodeSinkConnectorConfig;
-import org.apache.geode.cache.Region;
-import org.apache.geode.cache.RegionExistsException;
-import org.apache.geode.cache.client.ClientRegionShortcut;
-import org.apache.kafka.connect.sink.SinkRecord;
-import org.apache.kafka.connect.sink.SinkTask;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import geode.kafka.GeodeContext;
+import geode.kafka.GeodeSinkConnectorConfig;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.apache.kafka.connect.sink.SinkTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionExistsException;
+import org.apache.geode.cache.client.ClientRegionShortcut;
+
/**
* TODO javaDoc
@@ -37,109 +38,113 @@
*/
public class GeodeKafkaSinkTask extends SinkTask {
- private static final Logger logger = LoggerFactory.getLogger(GeodeKafkaSinkTask.class);
+ private static final Logger logger = LoggerFactory.getLogger(GeodeKafkaSinkTask.class);
- private GeodeContext geodeContext;
- private int taskId;
- private Map<String, List<String>> topicToRegions;
- private Map<String, Region> regionNameToRegion;
- private boolean nullValuesMeansRemove = true;
+ private GeodeContext geodeContext;
+ private int taskId;
+ private Map<String, List<String>> topicToRegions;
+ private Map<String, Region> regionNameToRegion;
+ private boolean nullValuesMeansRemove = true;
- /**
- * {@inheritDoc}
- */
- @Override
- public String version() {
- //TODO
- return "unknown";
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public String version() {
+ // TODO
+ return "unknown";
+ }
+
+ @Override
+ public void start(Map<String, String> props) {
+ try {
+ GeodeSinkConnectorConfig geodeConnectorConfig = new GeodeSinkConnectorConfig(props);
+ configure(geodeConnectorConfig);
+ geodeContext = new GeodeContext();
+ geodeContext.connectClient(geodeConnectorConfig.getLocatorHostPorts(),
+ geodeConnectorConfig.getSecurityClientAuthInit());
+ regionNameToRegion = createProxyRegions(topicToRegions.values());
+ } catch (Exception e) {
+ logger.error("Unable to start sink task", e);
+ throw e;
}
+ }
- @Override
- public void start(Map<String, String> props) {
- try {
- GeodeSinkConnectorConfig geodeConnectorConfig = new GeodeSinkConnectorConfig(props);
- configure(geodeConnectorConfig);
- geodeContext = new GeodeContext();
- geodeContext.connectClient(geodeConnectorConfig.getLocatorHostPorts(), geodeConnectorConfig.getSecurityClientAuthInit());
- regionNameToRegion = createProxyRegions(topicToRegions.values());
- } catch (Exception e) {
- logger.error("Unable to start sink task", e);
- throw e;
- }
+ void configure(GeodeSinkConnectorConfig geodeConnectorConfig) {
+ logger.debug("GeodeKafkaSourceTask id:" + geodeConnectorConfig.getTaskId() + " starting");
+ taskId = geodeConnectorConfig.getTaskId();
+ topicToRegions = geodeConnectorConfig.getTopicToRegions();
+ nullValuesMeansRemove = geodeConnectorConfig.getNullValuesMeanRemove();
+ }
+
+ // For tests only
+ void setRegionNameToRegion(Map<String, Region> regionNameToRegion) {
+ this.regionNameToRegion = regionNameToRegion;
+ }
+
+ @Override
+ public void put(Collection<SinkRecord> records) {
+ put(records, new HashMap());
+ }
+
+ void put(Collection<SinkRecord> records, Map<String, BatchRecords> batchRecordsMap) {
+ // spin off a new thread to handle this operation? Downside is ordering and retries...
+ for (SinkRecord record : records) {
+ updateBatchForRegionByTopic(record, batchRecordsMap);
}
+ batchRecordsMap.entrySet().stream().forEach((entry) -> {
+ String region = entry.getKey();
+ BatchRecords batchRecords = entry.getValue();
+ batchRecords.executeOperations(regionNameToRegion.get(region));
+ });
+ }
- void configure(GeodeSinkConnectorConfig geodeConnectorConfig) {
- logger.debug("GeodeKafkaSourceTask id:" + geodeConnectorConfig.getTaskId() + " starting");
- taskId = geodeConnectorConfig.getTaskId();
- topicToRegions = geodeConnectorConfig.getTopicToRegions();
- nullValuesMeansRemove = geodeConnectorConfig.getNullValuesMeanRemove();
+ private void updateBatchForRegionByTopic(SinkRecord sinkRecord,
+ Map<String, BatchRecords> batchRecordsMap) {
+ Collection<String> regionsToUpdate = topicToRegions.get(sinkRecord.topic());
+ for (String region : regionsToUpdate) {
+ updateBatchRecordsForRecord(sinkRecord, batchRecordsMap, region);
}
+ }
- //For tests only
- void setRegionNameToRegion(Map<String, Region> regionNameToRegion) {
- this.regionNameToRegion = regionNameToRegion;
+ private void updateBatchRecordsForRecord(SinkRecord record,
+ Map<String, BatchRecords> batchRecordsMap, String region) {
+ BatchRecords batchRecords = batchRecordsMap.get(region);
+ if (batchRecords == null) {
+ batchRecords = new BatchRecords();
+ batchRecordsMap.put(region, batchRecords);
}
-
- @Override
- public void put(Collection<SinkRecord> records) {
- put(records, new HashMap());
+ if (record.key() != null) {
+ if (record.value() == null && nullValuesMeansRemove) {
+ batchRecords.addRemoveOperation(record);
+ } else {
+ batchRecords.addUpdateOperation(record, nullValuesMeansRemove);
+ }
+ } else {
+ // Invest in configurable auto key generator?
+ logger.warn("Unable to push to Geode, missing key in record : " + record.value());
}
+ }
- void put(Collection<SinkRecord> records, Map<String, BatchRecords> batchRecordsMap) {
- //spin off a new thread to handle this operation? Downside is ordering and retries...
- for (SinkRecord record : records) {
- updateBatchForRegionByTopic(record, batchRecordsMap);
- }
- batchRecordsMap.entrySet().stream().forEach((entry) -> {
- String region = entry.getKey();
- BatchRecords batchRecords = entry.getValue();
- batchRecords.executeOperations(regionNameToRegion.get(region));
- });
+ private Map<String, Region> createProxyRegions(Collection<List<String>> regionNames) {
+ List<String> flat = regionNames.stream().flatMap(List::stream).collect(Collectors.toList());
+ return flat.stream().map(regionName -> createProxyRegion(regionName))
+ .collect(Collectors.toMap(region -> region.getName(), region -> region));
+ }
+
+ private Region createProxyRegion(String regionName) {
+ try {
+ return geodeContext.getClientCache().createClientRegionFactory(ClientRegionShortcut.PROXY)
+ .create(regionName);
+ } catch (RegionExistsException e) {
+ // Each task is a seperate parallel task controlled by kafka.
+ return geodeContext.getClientCache().getRegion(regionName);
}
+ }
- private void updateBatchForRegionByTopic(SinkRecord sinkRecord, Map<String, BatchRecords> batchRecordsMap) {
- Collection<String> regionsToUpdate = topicToRegions.get(sinkRecord.topic());
- for (String region : regionsToUpdate) {
- updateBatchRecordsForRecord(sinkRecord, batchRecordsMap, region);
- }
- }
+ @Override
+ public void stop() {
+ geodeContext.getClientCache().close(false);
+ }
- private void updateBatchRecordsForRecord(SinkRecord record, Map<String, BatchRecords> batchRecordsMap, String region) {
- BatchRecords batchRecords = batchRecordsMap.get(region);
- if (batchRecords == null) {
- batchRecords = new BatchRecords();
- batchRecordsMap.put(region, batchRecords);
- }
- if (record.key() != null) {
- if (record.value() == null && nullValuesMeansRemove) {
- batchRecords.addRemoveOperation(record);
- } else {
- batchRecords.addUpdateOperation(record, nullValuesMeansRemove);
- }
- } else {
- //Invest in configurable auto key generator?
- logger.warn("Unable to push to Geode, missing key in record : " + record.value());
- }
- }
-
- private Map<String, Region> createProxyRegions(Collection<List<String>> regionNames) {
- List<String> flat = regionNames.stream().flatMap(List::stream).collect(Collectors.toList());
- return flat.stream().map(regionName -> createProxyRegion(regionName)).collect(Collectors.toMap(region->region.getName(), region -> region));
- }
-
- private Region createProxyRegion(String regionName) {
- try {
- return geodeContext.getClientCache().createClientRegionFactory(ClientRegionShortcut.PROXY).create(regionName);
- }
- catch (RegionExistsException e) {
- //Each task is a seperate parallel task controlled by kafka.
- return geodeContext.getClientCache().getRegion(regionName);
- }
- }
-
- @Override
- public void stop() {
- geodeContext.getClientCache().close(false);
- }
-
-}
\ No newline at end of file
+}
diff --git a/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java b/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java
index 59a57cd..e416a9f 100644
--- a/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java
+++ b/src/main/java/geode/kafka/sink/GeodeSinkConnectorConfig.java
@@ -18,26 +18,26 @@
import java.util.Map;
public class GeodeSinkConnectorConfig extends GeodeConnectorConfig {
- //Used by sink
- public static final String TOPIC_TO_REGION_BINDINGS = "topicToRegions";
- public static final String NULL_VALUES_MEAN_REMOVE = "nullValuesMeanRemove";
- public static final String DEFAULT_NULL_VALUES_MEAN_REMOVE = "true";
+ // Used by sink
+ public static final String TOPIC_TO_REGION_BINDINGS = "topicToRegions";
+ public static final String NULL_VALUES_MEAN_REMOVE = "nullValuesMeanRemove";
+ public static final String DEFAULT_NULL_VALUES_MEAN_REMOVE = "true";
- private Map<String, List<String>> topicToRegions;
- private final boolean nullValuesMeanRemove;
+ private Map<String, List<String>> topicToRegions;
+ private final boolean nullValuesMeanRemove;
- public GeodeSinkConnectorConfig(Map<String, String> connectorProperties) {
- super(connectorProperties);
- topicToRegions = parseTopicToRegions(connectorProperties.get(TOPIC_TO_REGION_BINDINGS));
- nullValuesMeanRemove = Boolean.parseBoolean(connectorProperties.get(NULL_VALUES_MEAN_REMOVE));
- }
+ public GeodeSinkConnectorConfig(Map<String, String> connectorProperties) {
+ super(connectorProperties);
+ topicToRegions = parseTopicToRegions(connectorProperties.get(TOPIC_TO_REGION_BINDINGS));
+ nullValuesMeanRemove = Boolean.parseBoolean(connectorProperties.get(NULL_VALUES_MEAN_REMOVE));
+ }
- public Map<String, List<String>> getTopicToRegions() {
- return topicToRegions;
- }
+ public Map<String, List<String>> getTopicToRegions() {
+ return topicToRegions;
+ }
- public boolean getNullValuesMeanRemove() {
- return nullValuesMeanRemove;
- }
+ public boolean getNullValuesMeanRemove() {
+ return nullValuesMeanRemove;
+ }
}
diff --git a/src/main/java/geode/kafka/source/EventBufferSupplier.java b/src/main/java/geode/kafka/source/EventBufferSupplier.java
index d844744..8b05e22 100644
--- a/src/main/java/geode/kafka/source/EventBufferSupplier.java
+++ b/src/main/java/geode/kafka/source/EventBufferSupplier.java
@@ -1,3 +1,17 @@
+/*
+ * 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 geode.kafka.source;
import java.util.concurrent.BlockingQueue;
diff --git a/src/main/java/geode/kafka/source/GeodeEvent.java b/src/main/java/geode/kafka/source/GeodeEvent.java
index 3de8abd..f569f2b 100644
--- a/src/main/java/geode/kafka/source/GeodeEvent.java
+++ b/src/main/java/geode/kafka/source/GeodeEvent.java
@@ -21,19 +21,19 @@
*/
public class GeodeEvent {
- private String regionName;
- private CqEvent event;
+ private String regionName;
+ private CqEvent event;
- public GeodeEvent(String regionName, CqEvent event) {
- this.regionName = regionName;
- this.event = event;
- }
+ public GeodeEvent(String regionName, CqEvent event) {
+ this.regionName = regionName;
+ this.event = event;
+ }
- public String getRegionName() {
- return regionName;
- }
+ public String getRegionName() {
+ return regionName;
+ }
- public CqEvent getEvent() {
- return event;
- }
+ public CqEvent getEvent() {
+ return event;
+ }
}
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSource.java b/src/main/java/geode/kafka/source/GeodeKafkaSource.java
index b81401a..f150e07 100644
--- a/src/main/java/geode/kafka/source/GeodeKafkaSource.java
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSource.java
@@ -14,18 +14,6 @@
*/
package geode.kafka.source;
-import geode.kafka.GeodeConnectorConfig;
-import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.common.utils.AppInfoParser;
-import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.source.SourceConnector;
-import org.apache.kafka.connect.util.ConnectorUtils;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
import static geode.kafka.GeodeConnectorConfig.DEFAULT_LOCATOR;
import static geode.kafka.GeodeConnectorConfig.LOCATORS;
import static geode.kafka.source.GeodeSourceConnectorConfig.BATCH_SIZE;
@@ -43,11 +31,23 @@
import static geode.kafka.source.GeodeSourceConnectorConfig.QUEUE_SIZE;
import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import geode.kafka.GeodeConnectorConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.source.SourceConnector;
+import org.apache.kafka.connect.util.ConnectorUtils;
+
public class GeodeKafkaSource extends SourceConnector {
private Map<String, String> sharedProps;
- //TODO maybe club this into GeodeConnnectorConfig
+ // TODO maybe club this into GeodeConnnectorConfig
private static final ConfigDef CONFIG_DEF = new ConfigDef();
@@ -59,14 +59,16 @@
@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
List<Map<String, String>> taskConfigs = new ArrayList<>();
- List<String> bindings = GeodeConnectorConfig.parseStringByComma(sharedProps.get(REGION_TO_TOPIC_BINDINGS));
+ List<String> bindings =
+ GeodeConnectorConfig.parseStringByComma(sharedProps.get(REGION_TO_TOPIC_BINDINGS));
List<List<String>> bindingsPerTask = ConnectorUtils.groupPartitions(bindings, maxTasks);
for (int i = 0; i < maxTasks; i++) {
Map<String, String> taskProps = new HashMap<>();
taskProps.putAll(sharedProps);
taskProps.put(GeodeConnectorConfig.TASK_ID, "" + i);
- taskProps.put(CQS_TO_REGISTER, GeodeConnectorConfig.reconstructString(bindingsPerTask.get(i)));
+ taskProps.put(CQS_TO_REGISTER,
+ GeodeConnectorConfig.reconstructString(bindingsPerTask.get(i)));
taskConfigs.add(taskProps);
}
return taskConfigs;
@@ -84,7 +86,7 @@
}
private Map<String, String> computeMissingConfigurations(Map<String, String> props) {
- props.computeIfAbsent(LOCATORS, (key)-> DEFAULT_LOCATOR);
+ props.computeIfAbsent(LOCATORS, (key) -> DEFAULT_LOCATOR);
props.computeIfAbsent(DURABLE_CLIENT_TIME_OUT, (key) -> DEFAULT_DURABLE_CLIENT_TIMEOUT);
props.computeIfAbsent(DURABLE_CLIENT_ID_PREFIX, (key) -> DEFAULT_DURABLE_CLIENT_ID);
props.computeIfAbsent(BATCH_SIZE, (key) -> DEFAULT_BATCH_SIZE);
@@ -101,7 +103,7 @@
@Override
public String version() {
- //TODO
+ // TODO
return AppInfoParser.getVersion();
}
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java b/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
index 5c8a152..b0b8c6a 100644
--- a/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSourceListener.java
@@ -14,64 +14,66 @@
*/
package geode.kafka.source;
-import org.apache.geode.cache.query.CqEvent;
-import org.apache.geode.cache.query.CqStatusListener;
+import java.util.concurrent.TimeUnit;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.concurrent.TimeUnit;
+import org.apache.geode.cache.query.CqEvent;
+import org.apache.geode.cache.query.CqStatusListener;
class GeodeKafkaSourceListener implements CqStatusListener {
- private static final Logger logger = LoggerFactory.getLogger(GeodeKafkaSourceListener.class);
+ private static final Logger logger = LoggerFactory.getLogger(GeodeKafkaSourceListener.class);
- public String regionName;
- private EventBufferSupplier eventBufferSupplier;
- private boolean initialResultsLoaded;
+ public String regionName;
+ private EventBufferSupplier eventBufferSupplier;
+ private boolean initialResultsLoaded;
- public GeodeKafkaSourceListener(EventBufferSupplier eventBufferSupplier, String regionName) {
- this.regionName = regionName;
- this.eventBufferSupplier = eventBufferSupplier;
- initialResultsLoaded = false;
+ public GeodeKafkaSourceListener(EventBufferSupplier eventBufferSupplier, String regionName) {
+ this.regionName = regionName;
+ this.eventBufferSupplier = eventBufferSupplier;
+ initialResultsLoaded = false;
+ }
+
+ @Override
+ public void onEvent(CqEvent aCqEvent) {
+ while (!initialResultsLoaded) {
+ Thread.yield();
}
+ try {
+ eventBufferSupplier.get().offer(new GeodeEvent(regionName, aCqEvent), 2, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
- @Override
- public void onEvent(CqEvent aCqEvent) {
- while (!initialResultsLoaded) {
- Thread.yield();
- }
+ while (true) {
try {
- eventBufferSupplier.get().offer(new GeodeEvent(regionName, aCqEvent), 2, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
-
- while (true) {
- try {
- if (!eventBufferSupplier.get().offer(new GeodeEvent(regionName, aCqEvent), 2, TimeUnit.SECONDS))
- break;
- } catch (InterruptedException ex) {
- ex.printStackTrace();
- }
- logger.info("GeodeKafkaSource Queue is full");
- }
+ if (!eventBufferSupplier.get().offer(new GeodeEvent(regionName, aCqEvent), 2,
+ TimeUnit.SECONDS))
+ break;
+ } catch (InterruptedException ex) {
+ ex.printStackTrace();
}
+ logger.info("GeodeKafkaSource Queue is full");
+ }
}
+ }
- @Override
- public void onError(CqEvent aCqEvent) {
+ @Override
+ public void onError(CqEvent aCqEvent) {
- }
+ }
- @Override
- public void onCqDisconnected() {
- //we should probably redistribute or reconnect
- }
+ @Override
+ public void onCqDisconnected() {
+ // we should probably redistribute or reconnect
+ }
- @Override
- public void onCqConnected() {
+ @Override
+ public void onCqConnected() {
- }
+ }
- public void signalInitialResultsLoaded() {
- initialResultsLoaded = true;
- }
+ public void signalInitialResultsLoaded() {
+ initialResultsLoaded = true;
+ }
}
diff --git a/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java b/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
index 6efeb85..da2119a 100644
--- a/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
+++ b/src/main/java/geode/kafka/source/GeodeKafkaSourceTask.java
@@ -14,151 +14,163 @@
*/
package geode.kafka.source;
-import geode.kafka.GeodeContext;
-import org.apache.geode.cache.query.CqAttributes;
-import org.apache.geode.cache.query.CqAttributesFactory;
-import org.apache.geode.cache.query.CqEvent;
-import org.apache.kafka.connect.source.SourceRecord;
-import org.apache.kafka.connect.source.SourceTask;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static geode.kafka.source.GeodeSourceConnectorConfig.BATCH_SIZE;
+import static geode.kafka.source.GeodeSourceConnectorConfig.QUEUE_SIZE;
+import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_PARTITION;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;
-import static geode.kafka.source.GeodeSourceConnectorConfig.BATCH_SIZE;
-import static geode.kafka.source.GeodeSourceConnectorConfig.QUEUE_SIZE;
-import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_PARTITION;
+import geode.kafka.GeodeContext;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.source.SourceTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.geode.cache.query.CqAttributes;
+import org.apache.geode.cache.query.CqAttributesFactory;
+import org.apache.geode.cache.query.CqEvent;
public class GeodeKafkaSourceTask extends SourceTask {
- private static final Logger logger = LoggerFactory.getLogger(GeodeKafkaSourceTask.class);
+ private static final Logger logger = LoggerFactory.getLogger(GeodeKafkaSourceTask.class);
- private static final String TASK_PREFIX = "TASK";
- private static final String DOT = ".";
+ private static final String TASK_PREFIX = "TASK";
+ private static final String DOT = ".";
- //property string to pass in to identify task
- private static final Map<String, Long> OFFSET_DEFAULT = createOffset();
+ // property string to pass in to identify task
+ private static final Map<String, Long> OFFSET_DEFAULT = createOffset();
- private GeodeContext geodeContext;
- private GeodeSourceConnectorConfig geodeConnectorConfig;
- private EventBufferSupplier eventBufferSupplier;
- private Map<String, List<String>> regionToTopics;
- private Map<String, Map<String, String>> sourcePartitions;
- private int batchSize;
+ private GeodeContext geodeContext;
+ private GeodeSourceConnectorConfig geodeConnectorConfig;
+ private EventBufferSupplier eventBufferSupplier;
+ private Map<String, List<String>> regionToTopics;
+ private Map<String, Map<String, String>> sourcePartitions;
+ private int batchSize;
- private static Map<String, Long> createOffset() {
- Map<String, Long> offset = new HashMap<>();
- offset.put("OFFSET", 0L);
- return offset;
+ private static Map<String, Long> createOffset() {
+ Map<String, Long> offset = new HashMap<>();
+ offset.put("OFFSET", 0L);
+ return offset;
+ }
+
+ @Override
+ public String version() {
+ return null;
+ }
+
+ @Override
+ public void start(Map<String, String> props) {
+ try {
+ geodeConnectorConfig = new GeodeSourceConnectorConfig(props);
+ int taskId = geodeConnectorConfig.getTaskId();
+ logger.debug("GeodeKafkaSourceTask id:" + geodeConnectorConfig.getTaskId() + " starting");
+ geodeContext = new GeodeContext();
+ geodeContext.connectClient(geodeConnectorConfig.getLocatorHostPorts(),
+ geodeConnectorConfig.getDurableClientId(), geodeConnectorConfig.getDurableClientTimeout(),
+ geodeConnectorConfig.getSecurityClientAuthInit());
+
+ batchSize = Integer.parseInt(props.get(BATCH_SIZE));
+ eventBufferSupplier = new SharedEventBufferSupplier(Integer.parseInt(props.get(QUEUE_SIZE)));
+
+ regionToTopics = geodeConnectorConfig.getRegionToTopics();
+ geodeConnectorConfig.getCqsToRegister();
+ sourcePartitions = createSourcePartitionsMap(regionToTopics.keySet());
+
+ String cqPrefix = geodeConnectorConfig.getCqPrefix();
+ boolean loadEntireRegion = geodeConnectorConfig.getLoadEntireRegion();
+ installOnGeode(geodeConnectorConfig, geodeContext, eventBufferSupplier, cqPrefix,
+ loadEntireRegion);
+ } catch (Exception e) {
+ e.printStackTrace();
+ logger.error("Unable to start source task", e);
+ throw e;
}
+ }
- @Override
- public String version() {
- return null;
- }
-
- @Override
- public void start(Map<String, String> props) {
- try {
- geodeConnectorConfig = new GeodeSourceConnectorConfig(props);
- int taskId = geodeConnectorConfig.getTaskId();
- logger.debug("GeodeKafkaSourceTask id:" + geodeConnectorConfig.getTaskId() + " starting");
- geodeContext = new GeodeContext();
- geodeContext.connectClient(geodeConnectorConfig.getLocatorHostPorts(), geodeConnectorConfig.getDurableClientId(), geodeConnectorConfig.getDurableClientTimeout(), geodeConnectorConfig.getSecurityClientAuthInit());
-
- batchSize = Integer.parseInt(props.get(BATCH_SIZE));
- eventBufferSupplier = new SharedEventBufferSupplier(Integer.parseInt(props.get(QUEUE_SIZE)));
-
- regionToTopics = geodeConnectorConfig.getRegionToTopics();
- geodeConnectorConfig.getCqsToRegister();
- sourcePartitions = createSourcePartitionsMap(regionToTopics.keySet());
-
- String cqPrefix = geodeConnectorConfig.getCqPrefix();
- boolean loadEntireRegion = geodeConnectorConfig.getLoadEntireRegion();
- installOnGeode(geodeConnectorConfig, geodeContext, eventBufferSupplier, cqPrefix, loadEntireRegion);
- } catch (Exception e) {
- e.printStackTrace();
- logger.error("Unable to start source task", e);
- throw e;
+ @Override
+ public List<SourceRecord> poll() throws InterruptedException {
+ ArrayList<SourceRecord> records = new ArrayList<>(batchSize);
+ ArrayList<GeodeEvent> events = new ArrayList<>(batchSize);
+ if (eventBufferSupplier.get().drainTo(events, batchSize) > 0) {
+ for (GeodeEvent event : events) {
+ String regionName = event.getRegionName();
+ List<String> topics = regionToTopics.get(regionName);
+ for (String topic : topics) {
+ records.add(new SourceRecord(sourcePartitions.get(regionName), OFFSET_DEFAULT, topic,
+ null, event.getEvent().getKey(), null, event.getEvent().getNewValue()));
}
+ }
+ return records;
}
- @Override
- public List<SourceRecord> poll() throws InterruptedException {
- ArrayList<SourceRecord> records = new ArrayList<>(batchSize);
- ArrayList<GeodeEvent> events = new ArrayList<>(batchSize);
- if (eventBufferSupplier.get().drainTo(events, batchSize) > 0) {
- for (GeodeEvent event : events) {
- String regionName = event.getRegionName();
- List<String> topics = regionToTopics.get(regionName);
- for (String topic : topics) {
- records.add(new SourceRecord(sourcePartitions.get(regionName), OFFSET_DEFAULT, topic, null, event.getEvent().getKey(), null, event.getEvent().getNewValue()));
- }
- }
- return records;
- }
+ return null;
+ }
- return null;
- }
+ @Override
+ public void stop() {
+ geodeContext.getClientCache().close(true);
+ }
- @Override
- public void stop() {
- geodeContext.getClientCache().close(true);
+ void installOnGeode(GeodeSourceConnectorConfig geodeConnectorConfig, GeodeContext geodeContext,
+ EventBufferSupplier eventBuffer, String cqPrefix, boolean loadEntireRegion) {
+ boolean isDurable = geodeConnectorConfig.isDurable();
+ int taskId = geodeConnectorConfig.getTaskId();
+ for (String region : geodeConnectorConfig.getCqsToRegister()) {
+ installListenersToRegion(geodeContext, taskId, eventBuffer, region, cqPrefix,
+ loadEntireRegion, isDurable);
}
+ if (isDurable) {
+ geodeContext.getClientCache().readyForEvents();
+ }
+ }
- void installOnGeode(GeodeSourceConnectorConfig geodeConnectorConfig, GeodeContext geodeContext, EventBufferSupplier eventBuffer, String cqPrefix, boolean loadEntireRegion) {
- boolean isDurable = geodeConnectorConfig.isDurable();
- int taskId = geodeConnectorConfig.getTaskId();
- for (String region : geodeConnectorConfig.getCqsToRegister()) {
- installListenersToRegion(geodeContext, taskId, eventBuffer, region, cqPrefix, loadEntireRegion, isDurable);
- }
- if (isDurable) {
- geodeContext.getClientCache().readyForEvents();
- }
+ GeodeKafkaSourceListener installListenersToRegion(GeodeContext geodeContext, int taskId,
+ EventBufferSupplier eventBuffer, String regionName, String cqPrefix, boolean loadEntireRegion,
+ boolean isDurable) {
+ CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
+ GeodeKafkaSourceListener listener = new GeodeKafkaSourceListener(eventBuffer, regionName);
+ cqAttributesFactory.addCqListener(listener);
+ CqAttributes cqAttributes = cqAttributesFactory.create();
+ try {
+ if (loadEntireRegion) {
+ Collection<CqEvent> events =
+ geodeContext.newCqWithInitialResults(generateCqName(taskId, cqPrefix, regionName),
+ "select * from /" + regionName, cqAttributes,
+ isDurable);
+ eventBuffer.get().addAll(
+ events.stream().map(e -> new GeodeEvent(regionName, e)).collect(Collectors.toList()));
+ } else {
+ geodeContext.newCq(generateCqName(taskId, cqPrefix, regionName),
+ "select * from /" + regionName, cqAttributes,
+ isDurable);
+ }
+ } finally {
+ listener.signalInitialResultsLoaded();
}
+ return listener;
+ }
- GeodeKafkaSourceListener installListenersToRegion(GeodeContext geodeContext, int taskId, EventBufferSupplier eventBuffer, String regionName, String cqPrefix, boolean loadEntireRegion, boolean isDurable) {
- CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
- GeodeKafkaSourceListener listener = new GeodeKafkaSourceListener(eventBuffer, regionName);
- cqAttributesFactory.addCqListener(listener);
- CqAttributes cqAttributes = cqAttributesFactory.create();
- try {
- if (loadEntireRegion) {
- Collection<CqEvent> events = geodeContext.newCqWithInitialResults(generateCqName(taskId, cqPrefix, regionName), "select * from /" + regionName, cqAttributes,
- isDurable);
- eventBuffer.get().addAll(events.stream().map(e -> new GeodeEvent(regionName, e)).collect(Collectors.toList()));
- } else {
- geodeContext.newCq(generateCqName(taskId, cqPrefix, regionName), "select * from /" + regionName, cqAttributes,
- isDurable);
- }
- } finally {
- listener.signalInitialResultsLoaded();
- }
- return listener;
- }
+ /**
+ * converts a list of regions names into a map of source partitions
+ *
+ * @param regionNames list of regionNames
+ * @return Map<String, Map < String, String>> a map of source partitions, keyed by region name
+ */
+ Map<String, Map<String, String>> createSourcePartitionsMap(Collection<String> regionNames) {
+ return regionNames.stream().map(regionName -> {
+ Map<String, String> sourcePartition = new HashMap<>();
+ sourcePartition.put(REGION_PARTITION, regionName);
+ return sourcePartition;
+ }).collect(Collectors.toMap(s -> s.get(REGION_PARTITION), s -> s));
+ }
- /**
- * converts a list of regions names into a map of source partitions
- *
- * @param regionNames list of regionNames
- * @return Map<String, Map < String, String>> a map of source partitions, keyed by region name
- */
- Map<String, Map<String, String>> createSourcePartitionsMap(Collection<String> regionNames) {
- return regionNames.stream().map(regionName -> {
- Map<String, String> sourcePartition = new HashMap<>();
- sourcePartition.put(REGION_PARTITION, regionName);
- return sourcePartition;
- }).collect(Collectors.toMap(s -> s.get(REGION_PARTITION), s -> s));
- }
-
- String generateCqName(int taskId, String cqPrefix, String regionName) {
- return cqPrefix + DOT + TASK_PREFIX + taskId + DOT + regionName;
- }
+ String generateCqName(int taskId, String cqPrefix, String regionName) {
+ return cqPrefix + DOT + TASK_PREFIX + taskId + DOT + regionName;
+ }
}
diff --git a/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java b/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java
index 0b46d2b..4f0393d 100644
--- a/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java
+++ b/src/main/java/geode/kafka/source/GeodeSourceConnectorConfig.java
@@ -14,101 +14,101 @@
*/
package geode.kafka.source;
-import geode.kafka.GeodeConnectorConfig;
-
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import geode.kafka.GeodeConnectorConfig;
+
public class GeodeSourceConnectorConfig extends GeodeConnectorConfig {
- //Geode Configuration
- public static final String DURABLE_CLIENT_ID_PREFIX = "durableClientIdPrefix";
- public static final String DEFAULT_DURABLE_CLIENT_ID = "";
- public static final String DURABLE_CLIENT_TIME_OUT = "durableClientTimeout";
- public static final String DEFAULT_DURABLE_CLIENT_TIMEOUT = "60000";
+ // Geode Configuration
+ public static final String DURABLE_CLIENT_ID_PREFIX = "durableClientIdPrefix";
+ public static final String DEFAULT_DURABLE_CLIENT_ID = "";
+ public static final String DURABLE_CLIENT_TIME_OUT = "durableClientTimeout";
+ public static final String DEFAULT_DURABLE_CLIENT_TIMEOUT = "60000";
- public static final String CQ_PREFIX = "cqPrefix";
- public static final String DEFAULT_CQ_PREFIX = "cqForGeodeKafka";
+ public static final String CQ_PREFIX = "cqPrefix";
+ public static final String DEFAULT_CQ_PREFIX = "cqForGeodeKafka";
- /**
- * Used as a key for source partitions
- */
- public static final String REGION_PARTITION = "regionPartition";
- public static final String REGION_TO_TOPIC_BINDINGS = "regionToTopics";
- public static final String CQS_TO_REGISTER = "cqsToRegister";
+ /**
+ * Used as a key for source partitions
+ */
+ public static final String REGION_PARTITION = "regionPartition";
+ public static final String REGION_TO_TOPIC_BINDINGS = "regionToTopics";
+ public static final String CQS_TO_REGISTER = "cqsToRegister";
- public static final String BATCH_SIZE = "geodeConnectorBatchSize";
- public static final String DEFAULT_BATCH_SIZE = "100";
+ public static final String BATCH_SIZE = "geodeConnectorBatchSize";
+ public static final String DEFAULT_BATCH_SIZE = "100";
- public static final String QUEUE_SIZE = "geodeConnectorQueueSize";
- public static final String DEFAULT_QUEUE_SIZE = "10000";
+ public static final String QUEUE_SIZE = "geodeConnectorQueueSize";
+ public static final String DEFAULT_QUEUE_SIZE = "10000";
- public static final String LOAD_ENTIRE_REGION = "loadEntireRegion";
- public static final String DEFAULT_LOAD_ENTIRE_REGION = "false";
+ public static final String LOAD_ENTIRE_REGION = "loadEntireRegion";
+ public static final String DEFAULT_LOAD_ENTIRE_REGION = "false";
- private final String durableClientId;
- private final String durableClientIdPrefix;
- private final String durableClientTimeout;
- private final String cqPrefix;
- private final boolean loadEntireRegion;
+ private final String durableClientId;
+ private final String durableClientIdPrefix;
+ private final String durableClientTimeout;
+ private final String cqPrefix;
+ private final boolean loadEntireRegion;
- private Map<String, List<String>> regionToTopics;
- private Collection<String> cqsToRegister;
+ private Map<String, List<String>> regionToTopics;
+ private Collection<String> cqsToRegister;
- public GeodeSourceConnectorConfig(Map<String, String> connectorProperties) {
- super(connectorProperties);
- cqsToRegister = parseRegionToTopics(connectorProperties.get(CQS_TO_REGISTER)).keySet();
- regionToTopics = parseRegionToTopics(connectorProperties.get(REGION_TO_TOPIC_BINDINGS));
- durableClientIdPrefix = connectorProperties.get(DURABLE_CLIENT_ID_PREFIX);
- if (isDurable(durableClientIdPrefix)) {
- durableClientId = durableClientIdPrefix + taskId;
- } else {
- durableClientId = "";
- }
- durableClientTimeout = connectorProperties.get(DURABLE_CLIENT_TIME_OUT);
- cqPrefix = connectorProperties.get(CQ_PREFIX);
- loadEntireRegion = Boolean.parseBoolean(connectorProperties.get(LOAD_ENTIRE_REGION));
+ public GeodeSourceConnectorConfig(Map<String, String> connectorProperties) {
+ super(connectorProperties);
+ cqsToRegister = parseRegionToTopics(connectorProperties.get(CQS_TO_REGISTER)).keySet();
+ regionToTopics = parseRegionToTopics(connectorProperties.get(REGION_TO_TOPIC_BINDINGS));
+ durableClientIdPrefix = connectorProperties.get(DURABLE_CLIENT_ID_PREFIX);
+ if (isDurable(durableClientIdPrefix)) {
+ durableClientId = durableClientIdPrefix + taskId;
+ } else {
+ durableClientId = "";
}
+ durableClientTimeout = connectorProperties.get(DURABLE_CLIENT_TIME_OUT);
+ cqPrefix = connectorProperties.get(CQ_PREFIX);
+ loadEntireRegion = Boolean.parseBoolean(connectorProperties.get(LOAD_ENTIRE_REGION));
+ }
- public boolean isDurable() {
- return isDurable(durableClientId);
- }
+ public boolean isDurable() {
+ return isDurable(durableClientId);
+ }
- /**
- * @param durableClientId or prefix can be passed in. Either both will be "" or both will have a value
- * @return
- */
- boolean isDurable(String durableClientId) {
- return !durableClientId.equals("");
- }
+ /**
+ * @param durableClientId or prefix can be passed in. Either both will be "" or both will have a
+ * value
+ */
+ boolean isDurable(String durableClientId) {
+ return !durableClientId.equals("");
+ }
- public int getTaskId() {
- return taskId;
- }
+ public int getTaskId() {
+ return taskId;
+ }
- public String getDurableClientId() {
- return durableClientId;
- }
+ public String getDurableClientId() {
+ return durableClientId;
+ }
- public String getDurableClientTimeout() {
- return durableClientTimeout;
- }
+ public String getDurableClientTimeout() {
+ return durableClientTimeout;
+ }
- public String getCqPrefix() {
- return cqPrefix;
- }
+ public String getCqPrefix() {
+ return cqPrefix;
+ }
- public boolean getLoadEntireRegion() {
- return loadEntireRegion;
- }
+ public boolean getLoadEntireRegion() {
+ return loadEntireRegion;
+ }
- public Map<String, List<String>> getRegionToTopics() {
- return regionToTopics;
- }
+ public Map<String, List<String>> getRegionToTopics() {
+ return regionToTopics;
+ }
- public Collection<String> getCqsToRegister() {
- return cqsToRegister;
- }
+ public Collection<String> getCqsToRegister() {
+ return cqsToRegister;
+ }
}
diff --git a/src/main/java/geode/kafka/source/SharedEventBufferSupplier.java b/src/main/java/geode/kafka/source/SharedEventBufferSupplier.java
index 2a4c883..6ac6bb6 100644
--- a/src/main/java/geode/kafka/source/SharedEventBufferSupplier.java
+++ b/src/main/java/geode/kafka/source/SharedEventBufferSupplier.java
@@ -1,37 +1,51 @@
+/*
+ * 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 geode.kafka.source;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.function.Supplier;
public class SharedEventBufferSupplier implements EventBufferSupplier {
- private static BlockingQueue<GeodeEvent> eventBuffer;
+ private static BlockingQueue<GeodeEvent> eventBuffer;
- public SharedEventBufferSupplier(int size) {
- recreateEventBufferIfNeeded(size);
- }
+ public SharedEventBufferSupplier(int size) {
+ recreateEventBufferIfNeeded(size);
+ }
- BlockingQueue recreateEventBufferIfNeeded(int size) {
+ BlockingQueue recreateEventBufferIfNeeded(int size) {
+ if (eventBuffer == null || (eventBuffer.size() + eventBuffer.remainingCapacity()) != size) {
+ synchronized (GeodeKafkaSource.class) {
if (eventBuffer == null || (eventBuffer.size() + eventBuffer.remainingCapacity()) != size) {
- synchronized (GeodeKafkaSource.class) {
- if (eventBuffer == null || (eventBuffer.size() + eventBuffer.remainingCapacity()) != size) {
- BlockingQueue<GeodeEvent> oldEventBuffer = eventBuffer;
- eventBuffer = new LinkedBlockingQueue<>(size);
- if (oldEventBuffer != null) {
- eventBuffer.addAll(oldEventBuffer);
- }
- }
- }
+ BlockingQueue<GeodeEvent> oldEventBuffer = eventBuffer;
+ eventBuffer = new LinkedBlockingQueue<>(size);
+ if (oldEventBuffer != null) {
+ eventBuffer.addAll(oldEventBuffer);
+ }
}
- return eventBuffer;
+ }
}
+ return eventBuffer;
+ }
- /**
- * Callers should not store a reference to this and instead always call get to make sure we always use the latest buffer
- * Buffers themselves shouldn't change often but in cases where we want to modify the size
- */
- public BlockingQueue<GeodeEvent> get() {
- return eventBuffer;
- }
+ /**
+ * Callers should not store a reference to this and instead always call get to make sure we always
+ * use the latest buffer
+ * Buffers themselves shouldn't change often but in cases where we want to modify the size
+ */
+ public BlockingQueue<GeodeEvent> get() {
+ return eventBuffer;
+ }
}
diff --git a/src/test/java/geode/kafka/GeodeConnectorConfigTest.java b/src/test/java/geode/kafka/GeodeConnectorConfigTest.java
index 84a3ec5..c6fe491 100644
--- a/src/test/java/geode/kafka/GeodeConnectorConfigTest.java
+++ b/src/test/java/geode/kafka/GeodeConnectorConfigTest.java
@@ -14,15 +14,6 @@
*/
package geode.kafka;
-import junitparams.JUnitParamsRunner;
-import junitparams.Parameters;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-
import static org.hamcrest.CoreMatchers.allOf;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
@@ -30,122 +21,130 @@
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
@RunWith(JUnitParamsRunner.class)
public class GeodeConnectorConfigTest {
- @Test
- public void parseRegionNamesShouldSplitOnComma() {
- GeodeConnectorConfig config = new GeodeConnectorConfig();
- List<String> regionNames = config.parseStringByComma("region1,region2,region3,region4");
- assertEquals(4, regionNames.size());
- assertThat(true, allOf(is(regionNames.contains("region1"))
- , is(regionNames.contains("region2"))
- , is(regionNames.contains("region3"))
- , is(regionNames.contains("region4"))));
+ @Test
+ public void parseRegionNamesShouldSplitOnComma() {
+ GeodeConnectorConfig config = new GeodeConnectorConfig();
+ List<String> regionNames = config.parseStringByComma("region1,region2,region3,region4");
+ assertEquals(4, regionNames.size());
+ assertThat(true, allOf(is(regionNames.contains("region1")), is(regionNames.contains("region2")),
+ is(regionNames.contains("region3")), is(regionNames.contains("region4"))));
+ }
+
+ @Test
+ public void parseRegionNamesShouldChomp() {
+ GeodeConnectorConfig config = new GeodeConnectorConfig();
+ List<String> regionNames = config.parseStringByComma("region1, region2, region3,region4");
+ assertEquals(4, regionNames.size());
+ assertThat(true,
+ allOf(is(regionNames instanceof List), is(regionNames.contains("region1")),
+ is(regionNames.contains("region2")), is(regionNames.contains("region3")),
+ is(regionNames.contains("region4"))));
+ }
+
+ @Test
+ public void shouldBeAbleToParseGeodeLocatorStrings() {
+ GeodeConnectorConfig config = new GeodeConnectorConfig();
+ String locatorString = "localhost[8888], localhost[8881]";
+ List<LocatorHostPort> locators = config.parseLocators(locatorString);
+ assertThat(2, is(locators.size()));
+ }
+
+ @Test
+ @Parameters(method = "oneToOneBindings")
+ public void parseBindingsCanSplitOneToOneBindings(String value) {
+ List<String> splitBindings = GeodeConnectorConfig.parseBindings(value);
+ assertEquals(2, splitBindings.size());
+ }
+
+ @Test
+ public void parseBindingsCanSplitASingleOneToOneBindings() {
+ String binding = "[region1:topic1]";
+ List<String> splitBindings = GeodeConnectorConfig.parseBindings(binding);
+ assertEquals(1, splitBindings.size());
+ assertEquals(binding.replaceAll("\\[", "").replaceAll("\\]", ""), splitBindings.get(0));
+ }
+
+ public List<String> oneToOneBindings() {
+ return Arrays.asList(
+ new String[] {"[region1:topic1],[region2:topic2]", "[region1:topic1] , [region2:topic2]",
+ "[region1:topic1], [region2:topic2] ,", "[region1: topic1], [region2 :topic2]"});
+ }
+
+ @Test
+ @Parameters(method = "oneToManyBindings")
+ public void parseBindingsCanSplitOneToManyBindings(String value) {
+ List<String> splitBindings = GeodeConnectorConfig.parseBindings(value);
+ assertEquals(Arrays.toString(splitBindings.toArray()), 2, splitBindings.size());
+ }
+
+ public List<String> oneToManyBindings() {
+ return Arrays.asList(new String[] {"[region1:topic1,topic2],[region2:topic2,topic3]",
+ "[region1:topic1 , topic2] , [region2:topic2 , topic3]",
+ "[region1:topic1 ,], [region2:topic2 ,] ,",
+ "[region1: topic1 ,topic3], [region2 :topic2]"});
+ }
+
+ @Test
+ @Parameters(method = "oneToManyBindings")
+ public void reconstructBindingsToStringShouldReformAParsableString(String value) {
+ List<String> splitBindings = GeodeConnectorConfig.parseBindings(value);
+ String reconstructString = GeodeConnectorConfig.reconstructString(splitBindings);
+ splitBindings = GeodeConnectorConfig.parseBindings(reconstructString);
+ assertEquals(Arrays.toString(splitBindings.toArray()), 2, splitBindings.size());
+ for (String topicOrRegion : splitBindings) {
+ assertFalse(topicOrRegion.contains("\\["));
+ assertFalse(topicOrRegion.contains("\\]"));
}
+ }
- @Test
- public void parseRegionNamesShouldChomp() {
- GeodeConnectorConfig config = new GeodeConnectorConfig();
- List<String> regionNames = config.parseStringByComma("region1, region2, region3,region4");
- assertEquals(4, regionNames.size());
- assertThat(true, allOf(is(regionNames instanceof List)
- , is(regionNames.contains("region1"))
- , is(regionNames.contains("region2"))
- , is(regionNames.contains("region3"))
- , is(regionNames.contains("region4"))));
- }
+ @Test
+ @Parameters(method = "oneToOneBindings")
+ public void configurationShouldReturnRegionToTopicsMappingWhenParseRegionToTopics(String value) {
+ Map<String, List<String>> regionToTopics = GeodeConnectorConfig.parseRegionToTopics(value);
+ assertEquals(2, regionToTopics.size());
+ assertTrue(regionToTopics.get("region1") != null);
+ assertEquals(1, regionToTopics.get("region1").size());
+ assertTrue(regionToTopics.get("region1").contains("topic1"));
+ }
- @Test
- public void shouldBeAbleToParseGeodeLocatorStrings() {
- GeodeConnectorConfig config = new GeodeConnectorConfig();
- String locatorString="localhost[8888], localhost[8881]";
- List<LocatorHostPort> locators = config.parseLocators(locatorString);
- assertThat(2, is(locators.size()));
- }
-
- @Test
- @Parameters(method="oneToOneBindings")
- public void parseBindingsCanSplitOneToOneBindings(String value) {
- List<String> splitBindings = GeodeConnectorConfig.parseBindings(value);
- assertEquals(2, splitBindings.size());
- }
-
- @Test
- public void parseBindingsCanSplitASingleOneToOneBindings() {
- String binding = "[region1:topic1]";
- List<String> splitBindings = GeodeConnectorConfig.parseBindings(binding);
- assertEquals(1, splitBindings.size());
- assertEquals(binding.replaceAll("\\[", "").replaceAll("\\]", ""), splitBindings.get(0));
- }
-
- public List<String> oneToOneBindings() {
- return Arrays.asList(new String[]{"[region1:topic1],[region2:topic2]"
- ,"[region1:topic1] , [region2:topic2]"
- ,"[region1:topic1], [region2:topic2] ,"
- ,"[region1: topic1], [region2 :topic2]"});
- }
-
- @Test
- @Parameters(method="oneToManyBindings")
- public void parseBindingsCanSplitOneToManyBindings(String value) {
- List<String> splitBindings = GeodeConnectorConfig.parseBindings(value);
- assertEquals(Arrays.toString(splitBindings.toArray()), 2, splitBindings.size());
- }
-
- public List<String> oneToManyBindings() {
- return Arrays.asList(new String[]{"[region1:topic1,topic2],[region2:topic2,topic3]"
- ,"[region1:topic1 , topic2] , [region2:topic2 , topic3]"
- ,"[region1:topic1 ,], [region2:topic2 ,] ,"
- ,"[region1: topic1 ,topic3], [region2 :topic2]"});
- }
-
- @Test
- @Parameters(method="oneToManyBindings")
- public void reconstructBindingsToStringShouldReformAParsableString(String value) {
- List<String> splitBindings = GeodeConnectorConfig.parseBindings(value);
- String reconstructString = GeodeConnectorConfig.reconstructString(splitBindings);
- splitBindings = GeodeConnectorConfig.parseBindings(reconstructString);
- assertEquals(Arrays.toString(splitBindings.toArray()), 2, splitBindings.size());
- for(String topicOrRegion: splitBindings) {
- assertFalse(topicOrRegion.contains("\\["));
- assertFalse(topicOrRegion.contains("\\]"));
- }
- }
-
- @Test
- @Parameters(method="oneToOneBindings")
- public void configurationShouldReturnRegionToTopicsMappingWhenParseRegionToTopics(String value) {
- Map<String, List<String>> regionToTopics = GeodeConnectorConfig.parseRegionToTopics(value);
- assertEquals(2, regionToTopics.size());
- assertTrue(regionToTopics.get("region1") != null);
- assertEquals(1, regionToTopics.get("region1").size());
- assertTrue(regionToTopics.get("region1").contains("topic1"));
- }
-
- @Test
- public void regionToTopicParsingShouldParseCorrectlyWithASingleBinding() {
- Map<String, List<String>> regionToTopics = GeodeConnectorConfig.parseRegionToTopics("[region1:topic1]");
- assertTrue(regionToTopics.get("region1") != null);
- assertEquals(1, regionToTopics.get("region1").size());
- assertTrue(regionToTopics.get("region1").contains("topic1"));
- }
+ @Test
+ public void regionToTopicParsingShouldParseCorrectlyWithASingleBinding() {
+ Map<String, List<String>> regionToTopics =
+ GeodeConnectorConfig.parseRegionToTopics("[region1:topic1]");
+ assertTrue(regionToTopics.get("region1") != null);
+ assertEquals(1, regionToTopics.get("region1").size());
+ assertTrue(regionToTopics.get("region1").contains("topic1"));
+ }
- /*
- taskId = Integer.parseInt(connectorProperties.get(TASK_ID));
- durableClientIdPrefix = connectorProperties.get(DURABLE_CLIENT_ID_PREFIX);
- if (isDurable(durableClientIdPrefix)) {
- durableClientId = durableClientIdPrefix + taskId;
- } else {
- durableClientId = "";
- }
- durableClientTimeout = connectorProperties.get(DURABLE_CLIENT_TIME_OUT);
- regionToTopics = parseRegionToTopics(connectorProperties.get(GeodeConnectorConfig.REGION_TO_TOPIC_BINDINGS));
- topicToRegions = parseTopicToRegions(connectorProperties.get(GeodeConnectorConfig.REGION_TO_TOPIC_BINDINGS));
- locatorHostPorts = parseLocators(connectorProperties.get(GeodeConnectorConfig.LOCATORS));
-
- */
+ /*
+ * taskId = Integer.parseInt(connectorProperties.get(TASK_ID));
+ * durableClientIdPrefix = connectorProperties.get(DURABLE_CLIENT_ID_PREFIX);
+ * if (isDurable(durableClientIdPrefix)) {
+ * durableClientId = durableClientIdPrefix + taskId;
+ * } else {
+ * durableClientId = "";
+ * }
+ * durableClientTimeout = connectorProperties.get(DURABLE_CLIENT_TIME_OUT);
+ * regionToTopics =
+ * parseRegionToTopics(connectorProperties.get(GeodeConnectorConfig.REGION_TO_TOPIC_BINDINGS));
+ * topicToRegions =
+ * parseTopicToRegions(connectorProperties.get(GeodeConnectorConfig.REGION_TO_TOPIC_BINDINGS));
+ * locatorHostPorts = parseLocators(connectorProperties.get(GeodeConnectorConfig.LOCATORS));
+ *
+ */
diff --git a/src/test/java/geode/kafka/GeodeKafkaTestCluster.java b/src/test/java/geode/kafka/GeodeKafkaTestCluster.java
index 121a978..3ee5e09 100644
--- a/src/test/java/geode/kafka/GeodeKafkaTestCluster.java
+++ b/src/test/java/geode/kafka/GeodeKafkaTestCluster.java
@@ -14,13 +14,20 @@
*/
package geode.kafka;
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
import kafka.admin.RackAwareMode;
import kafka.zk.AdminZkClient;
import kafka.zk.KafkaZkClient;
-import org.apache.geode.cache.Region;
-import org.apache.geode.cache.client.ClientCache;
-import org.apache.geode.cache.client.ClientCacheFactory;
-import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -33,27 +40,18 @@
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.WorkerConfig;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
-import org.junit.After;
import org.junit.AfterClass;
-import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import java.io.IOException;
-import java.time.Duration;
-import java.util.Collections;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static org.awaitility.Awaitility.await;
-import static org.junit.Assert.assertEquals;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.cache.client.ClientRegionShortcut;
public class GeodeKafkaTestCluster {
@@ -76,7 +74,8 @@
private static Consumer<String, String> consumer;
@BeforeClass
- public static void setup() throws IOException, QuorumPeerConfig.ConfigException, InterruptedException {
+ public static void setup()
+ throws IOException, QuorumPeerConfig.ConfigException, InterruptedException {
startZooKeeper();
startKafka();
startGeode();
@@ -86,11 +85,11 @@
@AfterClass
public static void shutdown() {
workerAndHerderCluster.stop();
- KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181",false,200000,
- 15000,10, Time.SYSTEM, "myGroup","myMetricType", null);
-// AdminZkClient adminZkClient = new AdminZkClient(zkClient);
-// adminZkClient.deleteTopic(TEST_TOPIC_FOR_SOURCE);
-// adminZkClient.deleteTopic(TEST_TOPIC_FOR_SINK);
+ KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181", false, 200000,
+ 15000, 10, Time.SYSTEM, "myGroup", "myMetricType", null);
+ // AdminZkClient adminZkClient = new AdminZkClient(zkClient);
+ // adminZkClient.deleteTopic(TEST_TOPIC_FOR_SOURCE);
+ // adminZkClient.deleteTopic(TEST_TOPIC_FOR_SINK);
zkClient.close();
kafkaLocalCluster.stop();
geodeLocalCluster.stop();
@@ -104,18 +103,19 @@
}
private static void createTopic(String topicName, int numPartitions, int replicationFactor) {
- KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181",false,200000,
- 15000,10, Time.SYSTEM, "myGroup","myMetricType", null);
+ KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181", false, 200000,
+ 15000, 10, Time.SYSTEM, "myGroup", "myMetricType", null);
Properties topicProperties = new Properties();
topicProperties.put("flush.messages", "1");
AdminZkClient adminZkClient = new AdminZkClient(zkClient);
- adminZkClient.createTopic(topicName, numPartitions,replicationFactor, topicProperties, RackAwareMode.Disabled$.MODULE$);
+ adminZkClient.createTopic(topicName, numPartitions, replicationFactor, topicProperties,
+ RackAwareMode.Disabled$.MODULE$);
}
private static void deleteTopic(String topicName) {
- KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181",false,200000,
- 15000,10, Time.SYSTEM, "myGroup","myMetricType", null);
+ KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181", false, 200000,
+ 15000, 10, Time.SYSTEM, "myGroup", "myMetricType", null);
AdminZkClient adminZkClient = new AdminZkClient(zkClient);
adminZkClient.deleteTopic(topicName);
}
@@ -129,7 +129,8 @@
zooKeeperLocalCluster.start();
}
- private static void startKafka() throws IOException, InterruptedException, QuorumPeerConfig.ConfigException {
+ private static void startKafka()
+ throws IOException, InterruptedException, QuorumPeerConfig.ConfigException {
kafkaLocalCluster = new KafkaLocalCluster(getKafkaConfig());
kafkaLocalCluster.start();
}
@@ -139,9 +140,10 @@
geodeLocalCluster.start();
}
- private static Properties getZooKeeperProperties() throws IOException {
+ private static Properties getZooKeeperProperties() throws IOException {
Properties properties = new Properties();
- properties.setProperty("dataDir", (debug)? "/tmp/zookeeper" :temporaryFolder.newFolder("zookeeper").getAbsolutePath());
+ properties.setProperty("dataDir",
+ (debug) ? "/tmp/zookeeper" : temporaryFolder.newFolder("zookeeper").getAbsolutePath());
properties.setProperty("clientPort", "2181");
properties.setProperty("tickTime", "2000");
return properties;
@@ -159,43 +161,43 @@
props.put("port", BROKER_PORT);
props.put("offsets.topic.replication.factor", "1");
- //Specifically GeodeKafka connector configs
+ // Specifically GeodeKafka connector configs
return props;
}
- //consumer props, less important, just for testing?
- public static Consumer<String,String> createConsumer() {
- final Properties props = new Properties();
+ // consumer props, less important, just for testing?
+ public static Consumer<String, String> createConsumer() {
+ final Properties props = new Properties();
props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG,
- "myGroup");
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
- StringDeserializer.class.getName());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
- StringDeserializer.class.getName());
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ "myGroup");
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+ StringDeserializer.class.getName());
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+ StringDeserializer.class.getName());
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// Create the consumer using props.
- final Consumer<String, String> consumer =
- new KafkaConsumer<>(props);
- // Subscribe to the topic.
- consumer.subscribe(Collections.singletonList(TEST_TOPIC_FOR_SOURCE));
- return consumer;
+ final Consumer<String, String> consumer =
+ new KafkaConsumer<>(props);
+ // Subscribe to the topic.
+ consumer.subscribe(Collections.singletonList(TEST_TOPIC_FOR_SOURCE));
+ return consumer;
}
- //consumer props, less important, just for testing?
- public static Producer<String,String> createProducer() {
+ // consumer props, less important, just for testing?
+ public static Producer<String, String> createProducer() {
final Properties props = new Properties();
props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
- StringSerializer.class.getName());
+ StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
- StringSerializer.class.getName());
+ StringSerializer.class.getName());
// Create the producer using props.
final Producer<String, String> producer =
- new KafkaProducer<>(props);
+ new KafkaProducer<>(props);
return producer;
}
@@ -207,9 +209,10 @@
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SOURCE);
+ Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY)
+ .create(TEST_REGION_FOR_SOURCE);
- for (int i = 0; i < 10 ; i++) {
+ for (int i = 0; i < 10; i++) {
region.put("KEY" + i, "VALUE" + i);
}
@@ -221,8 +224,7 @@
}
return valueReceived.get() == 10;
});
- }
- finally {
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SOURCE);
}
}
@@ -236,9 +238,10 @@
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SOURCE);
+ Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY)
+ .create(TEST_REGION_FOR_SOURCE);
- for (int i = 0; i < 10 ; i++) {
+ for (int i = 0; i < 10; i++) {
region.put("KEY" + i, "VALUE" + i);
}
@@ -250,23 +253,24 @@
}
return valueReceived.get() == 10;
});
- }
- finally {
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SOURCE);
}
}
@Test
- public void endToEndSourceSingleRegionMultiTaskMultiPartitionWithMoreTasksThanPartitionsTest() throws Exception {
+ public void endToEndSourceSingleRegionMultiTaskMultiPartitionWithMoreTasksThanPartitionsTest()
+ throws Exception {
try {
createTopic(TEST_TOPIC_FOR_SOURCE, 2, 1);
startWorker(5);
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SOURCE);
+ Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY)
+ .create(TEST_REGION_FOR_SOURCE);
- for (int i = 0; i < 10 ; i++) {
+ for (int i = 0; i < 10; i++) {
region.put("KEY" + i, "VALUE" + i);
}
@@ -278,8 +282,7 @@
}
return valueReceived.get() == 10;
});
- }
- finally {
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SOURCE);
}
}
@@ -292,7 +295,8 @@
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
+ Region region =
+ client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
Producer<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
@@ -300,23 +304,25 @@
}
int i = 0;
- await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
- }
- finally {
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SINK);
}
}
@Test
- public void endToEndWithOneTaskForASingleBindingAgainstAMultiPartitionTopicSinkTest() throws Exception {
+ public void endToEndWithOneTaskForASingleBindingAgainstAMultiPartitionTopicSinkTest()
+ throws Exception {
try {
createTopic(TEST_TOPIC_FOR_SINK, 10, 1);
startWorker(5);
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
+ Region region =
+ client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
Producer<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
@@ -324,22 +330,24 @@
}
int i = 0;
- await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
- }
- finally {
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SINK);
}
}
@Test
- public void endToEndWithOneTaskForASingleBindingAgainstAMultiPartitionTopicWithMoreWorkersSinkTest() throws Exception {
+ public void endToEndWithOneTaskForASingleBindingAgainstAMultiPartitionTopicWithMoreWorkersSinkTest()
+ throws Exception {
try {
createTopic(TEST_TOPIC_FOR_SINK, 10, 1);
startWorker(15);
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
+ Region region =
+ client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
Producer<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
@@ -347,9 +355,9 @@
}
int i = 0;
- await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
- }
- finally {
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SINK);
}
}
@@ -362,7 +370,8 @@
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
+ Region region =
+ client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
Producer<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
@@ -370,9 +379,9 @@
}
int i = 0;
- await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
- }
- finally {
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SINK);
}
}
@@ -385,17 +394,19 @@
consumer = createConsumer();
ClientCache client = createGeodeClient();
- Region region = client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
+ Region region =
+ client.createClientRegionFactory(ClientRegionShortcut.PROXY).create(TEST_REGION_FOR_SINK);
Producer<String, String> producer = createProducer();
for (int i = 0; i < 10; i++) {
- producer.send(new ProducerRecord(TEST_TOPIC_FOR_SINK, i, UUID.randomUUID().toString(), UUID.randomUUID().toString()));
+ producer.send(new ProducerRecord(TEST_TOPIC_FOR_SINK, i, UUID.randomUUID().toString(),
+ UUID.randomUUID().toString()));
}
int i = 0;
- await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
- }
- finally {
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(10, region.sizeOnServer()));
+ } finally {
deleteTopic(TEST_TOPIC_FOR_SINK);
}
}
diff --git a/src/test/java/geode/kafka/GeodeLocalCluster.java b/src/test/java/geode/kafka/GeodeLocalCluster.java
index 85b3a99..259d30b 100644
--- a/src/test/java/geode/kafka/GeodeLocalCluster.java
+++ b/src/test/java/geode/kafka/GeodeLocalCluster.java
@@ -18,25 +18,24 @@
public class GeodeLocalCluster {
- private JavaProcess locatorProcess;
- private JavaProcess serverProcess;
+ private JavaProcess locatorProcess;
+ private JavaProcess serverProcess;
- public GeodeLocalCluster() {
- locatorProcess = new JavaProcess(LocatorLauncherWrapper.class);
- serverProcess = new JavaProcess(ServerLauncherWrapper.class);
- }
+ public GeodeLocalCluster() {
+ locatorProcess = new JavaProcess(LocatorLauncherWrapper.class);
+ serverProcess = new JavaProcess(ServerLauncherWrapper.class);
+ }
- public void start() throws IOException, InterruptedException {
- System.out.println("starting locator");
- locatorProcess.exec("10334");
- Thread.sleep(15000);
- serverProcess.exec("40404");
- Thread.sleep(30000);
- }
+ public void start() throws IOException, InterruptedException {
+ System.out.println("starting locator");
+ locatorProcess.exec("10334");
+ Thread.sleep(15000);
+ serverProcess.exec("40404");
+ Thread.sleep(30000);
+ }
- public void stop() {
- serverProcess.destroy();
- locatorProcess.destroy();
- }
+ public void stop() {
+ serverProcess.destroy();
+ locatorProcess.destroy();
+ }
}
-
diff --git a/src/test/java/geode/kafka/JavaProcess.java b/src/test/java/geode/kafka/JavaProcess.java
index ed7a493..b130223 100644
--- a/src/test/java/geode/kafka/JavaProcess.java
+++ b/src/test/java/geode/kafka/JavaProcess.java
@@ -19,40 +19,41 @@
public class JavaProcess {
- public Process process;
- Class classWithMain;
+ public Process process;
+ Class classWithMain;
- public JavaProcess(Class classWithmain) {
- this.classWithMain = classWithmain;
+ public JavaProcess(Class classWithmain) {
+ this.classWithMain = classWithmain;
+ }
+
+ public void exec(String... args) throws IOException, InterruptedException {
+ String java =
+ System.getProperty("java.home") + File.separator + "bin" + File.separator + "java";
+ String classpath = System.getProperty("java.class.path");
+ String className = classWithMain.getName();
+
+ ProcessBuilder builder = new ProcessBuilder(
+ java, "-cp", classpath, className, convertArgsToString(args));
+
+ process = builder.inheritIO().start();
+ }
+
+ private String convertArgsToString(String... args) {
+ String string = "";
+ for (String arg : args) {
+ string += arg;
}
+ return string;
+ }
- public void exec(String... args) throws IOException, InterruptedException {
- String java = System.getProperty("java.home") + File.separator + "bin" + File.separator + "java";
- String classpath = System.getProperty("java.class.path");
- String className = classWithMain.getName();
+ public void waitFor() throws InterruptedException {
+ process.waitFor();
+ }
- ProcessBuilder builder = new ProcessBuilder(
- java, "-cp", classpath, className, convertArgsToString(args));
-
- process = builder.inheritIO().start();
- }
-
- private String convertArgsToString(String... args) {
- String string = "";
- for(String arg: args) {
- string += arg;
- }
- return string;
- }
-
- public void waitFor() throws InterruptedException {
- process.waitFor();
- }
-
- public void destroy() {
- process.destroy();
- }
+ public void destroy() {
+ process.destroy();
+ }
-}
\ No newline at end of file
+}
diff --git a/src/test/java/geode/kafka/KafkaLocalCluster.java b/src/test/java/geode/kafka/KafkaLocalCluster.java
index 8d09775..ef534d9 100644
--- a/src/test/java/geode/kafka/KafkaLocalCluster.java
+++ b/src/test/java/geode/kafka/KafkaLocalCluster.java
@@ -14,12 +14,12 @@
*/
package geode.kafka;
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaServerStartable;
-
import java.io.IOException;
import java.util.Properties;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServerStartable;
+
public class KafkaLocalCluster {
KafkaServerStartable kafka;
@@ -33,8 +33,7 @@
try {
kafka.startup();
System.out.println("Kafka started up");
- }
- catch (Throwable t) {
+ } catch (Throwable t) {
System.out.println(t);
}
}
diff --git a/src/test/java/geode/kafka/LocatorLauncherWrapper.java b/src/test/java/geode/kafka/LocatorLauncherWrapper.java
index 0bc446d..1b922c9 100644
--- a/src/test/java/geode/kafka/LocatorLauncherWrapper.java
+++ b/src/test/java/geode/kafka/LocatorLauncherWrapper.java
@@ -14,37 +14,38 @@
*/
package geode.kafka;
-import org.apache.geode.distributed.ConfigurationProperties;
-import org.apache.geode.distributed.Locator;
-
import java.io.File;
import java.io.IOException;
import java.util.Properties;
+import org.apache.geode.distributed.ConfigurationProperties;
+import org.apache.geode.distributed.Locator;
+
public class LocatorLauncherWrapper {
- public static void main(String[] args) throws IOException {
- Properties properties = new Properties();
-// String statsFile = new File(context.getOutputDir(), "stats.gfs").getAbsolutePath();
-// properties.setProperty(ConfigurationPropert/**/ies.STATISTIC_ARCHIVE_FILE, statsFile);
- properties.setProperty(ConfigurationProperties.NAME, "locator1");
+ public static void main(String[] args) throws IOException {
+ Properties properties = new Properties();
+ // String statsFile = new File(context.getOutputDir(), "stats.gfs").getAbsolutePath();
+ // properties.setProperty(ConfigurationPropert/**/ies.STATISTIC_ARCHIVE_FILE, statsFile);
+ properties.setProperty(ConfigurationProperties.NAME, "locator1");
- Locator.startLocatorAndDS(10334, new File("/Users/jhuynh/Pivotal/geode-kafka-connector/locator.log"), properties);
- while (true) {
-
- }
-//
-// LocatorLauncher locatorLauncher = new LocatorLauncher.Builder()
-// .setMemberName("locator1")
-//// .setPort(Integer.valueOf(args[0]))
-//// .setBindAddress("localhost")
-// .build();
-//
-// locatorLauncher.start();
-// while (!locatorLauncher.isRunning()) {
-//
-// }
-// System.out.println(locatorLauncher.getBindAddress() + ":" + locatorLauncher.getPort());
+ Locator.startLocatorAndDS(10334,
+ new File("/Users/jhuynh/Pivotal/geode-kafka-connector/locator.log"), properties);
+ while (true) {
}
+ //
+ // LocatorLauncher locatorLauncher = new LocatorLauncher.Builder()
+ // .setMemberName("locator1")
+ //// .setPort(Integer.valueOf(args[0]))
+ //// .setBindAddress("localhost")
+ // .build();
+ //
+ // locatorLauncher.start();
+ // while (!locatorLauncher.isRunning()) {
+ //
+ // }
+ // System.out.println(locatorLauncher.getBindAddress() + ":" + locatorLauncher.getPort());
+
+ }
}
diff --git a/src/test/java/geode/kafka/ServerLauncherWrapper.java b/src/test/java/geode/kafka/ServerLauncherWrapper.java
index 5204f65..f60af27 100644
--- a/src/test/java/geode/kafka/ServerLauncherWrapper.java
+++ b/src/test/java/geode/kafka/ServerLauncherWrapper.java
@@ -14,58 +14,58 @@
*/
package geode.kafka;
+import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_FOR_SINK;
+import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_FOR_SOURCE;
+
+import java.io.IOException;
+import java.util.Properties;
+
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.ConfigurationProperties;
-import java.io.IOException;
-import java.util.Properties;
-
-import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_FOR_SINK;
-import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_FOR_SOURCE;
-
public class ServerLauncherWrapper {
- public static void main(String... args) throws IOException {
-// ServerLauncher serverLauncher = new ServerLauncher.Builder()
-// .setMemberName("server1")
-//// .setServerPort(Integer.valueOf(args[0]))
-//// .setServerBindAddress("localhost")
-// // .set("locators", "localhost[10334]")
-//// .set("jmx-manager", "true")
-//// .set("jmx-manager-start", "true")
-// .build();
-//
-// serverLauncher.start();
-// System.out.println("Geode Server Launcher complete");
+ public static void main(String... args) throws IOException {
+ // ServerLauncher serverLauncher = new ServerLauncher.Builder()
+ // .setMemberName("server1")
+ //// .setServerPort(Integer.valueOf(args[0]))
+ //// .setServerBindAddress("localhost")
+ // // .set("locators", "localhost[10334]")
+ //// .set("jmx-manager", "true")
+ //// .set("jmx-manager-start", "true")
+ // .build();
+ //
+ // serverLauncher.start();
+ // System.out.println("Geode Server Launcher complete");
+ Properties properties = new Properties();
+ String locatorString = "localhost[10334]";
+ // String statsFile = new File(context.getOutputDir(), "stats.gfs").getAbsolutePath();
+ Cache cache = new CacheFactory(properties)
+ // .setPdxSerializer(new ReflectionBasedAutoSerializer("benchmark.geode.data.*"))
+ .set(ConfigurationProperties.LOCATORS, locatorString)
+ .set(ConfigurationProperties.NAME,
+ "server-1")
+ .set(ConfigurationProperties.LOG_FILE,
+ "/Users/jhuynh/Pivotal/geode-kafka-connector/server.log")
+ .set(ConfigurationProperties.LOG_LEVEL, "info")
+ // .set(ConfigurationProperties.STATISTIC_ARCHIVE_FILE, statsFile)
+ .create();
+ CacheServer cacheServer = cache.addCacheServer();
+ cacheServer.setPort(0);
+ cacheServer.start();
- Properties properties = new Properties();
- String locatorString = "localhost[10334]";
-// String statsFile = new File(context.getOutputDir(), "stats.gfs").getAbsolutePath();
- Cache cache = new CacheFactory(properties)
-// .setPdxSerializer(new ReflectionBasedAutoSerializer("benchmark.geode.data.*"))
- .set(ConfigurationProperties.LOCATORS, locatorString)
- .set(ConfigurationProperties.NAME,
- "server-1")
- .set(ConfigurationProperties.LOG_FILE, "/Users/jhuynh/Pivotal/geode-kafka-connector/server.log")
- .set(ConfigurationProperties.LOG_LEVEL, "info")
-// .set(ConfigurationProperties.STATISTIC_ARCHIVE_FILE, statsFile)
- .create();
- CacheServer cacheServer = cache.addCacheServer();
- cacheServer.setPort(0);
- cacheServer.start();
+ // create the region
+ cache.createRegionFactory(RegionShortcut.PARTITION).create(TEST_REGION_FOR_SINK);
+ cache.createRegionFactory(RegionShortcut.PARTITION).create(TEST_REGION_FOR_SOURCE);
+ System.out.println("starting cacheserver");
+ while (true) {
- //create the region
- cache.createRegionFactory(RegionShortcut.PARTITION).create(TEST_REGION_FOR_SINK);
- cache.createRegionFactory(RegionShortcut.PARTITION).create(TEST_REGION_FOR_SOURCE);
- System.out.println("starting cacheserver");
- while (true) {
-
- }
}
+ }
}
diff --git a/src/test/java/geode/kafka/WorkerAndHerderCluster.java b/src/test/java/geode/kafka/WorkerAndHerderCluster.java
index c388a5a..70461a3 100644
--- a/src/test/java/geode/kafka/WorkerAndHerderCluster.java
+++ b/src/test/java/geode/kafka/WorkerAndHerderCluster.java
@@ -18,19 +18,18 @@
public class WorkerAndHerderCluster {
- private JavaProcess workerAndHerder;
+ private JavaProcess workerAndHerder;
- public WorkerAndHerderCluster() {
- workerAndHerder = new JavaProcess(WorkerAndHerderWrapper.class);
- }
+ public WorkerAndHerderCluster() {
+ workerAndHerder = new JavaProcess(WorkerAndHerderWrapper.class);
+ }
- public void start(String maxTasks) throws IOException, InterruptedException {
- workerAndHerder.exec(maxTasks);
+ public void start(String maxTasks) throws IOException, InterruptedException {
+ workerAndHerder.exec(maxTasks);
- }
+ }
- public void stop() {
- workerAndHerder.destroy();
- }
+ public void stop() {
+ workerAndHerder.destroy();
+ }
}
-
diff --git a/src/test/java/geode/kafka/WorkerAndHerderWrapper.java b/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
index 24427aa..5f86985 100644
--- a/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
+++ b/src/test/java/geode/kafka/WorkerAndHerderWrapper.java
@@ -14,6 +14,16 @@
*/
package geode.kafka;
+import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_TO_TOPIC_BINDINGS;
+import static geode.kafka.GeodeKafkaTestCluster.TEST_TOPIC_FOR_SINK;
+import static geode.kafka.GeodeKafkaTestCluster.TEST_TOPIC_TO_REGION_BINDINGS;
+import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
import geode.kafka.sink.GeodeKafkaSink;
import geode.kafka.source.GeodeKafkaSource;
import org.apache.kafka.common.utils.SystemTime;
@@ -28,68 +38,62 @@
import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
import org.apache.kafka.connect.util.ConnectUtils;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import static geode.kafka.GeodeKafkaTestCluster.TEST_REGION_TO_TOPIC_BINDINGS;
-import static geode.kafka.GeodeKafkaTestCluster.TEST_TOPIC_FOR_SINK;
-import static geode.kafka.GeodeKafkaTestCluster.TEST_TOPIC_TO_REGION_BINDINGS;
-import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
-import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
-
public class WorkerAndHerderWrapper {
- public static void main(String[] args) throws IOException {
- String maxTasks = args[0];
+ public static void main(String[] args) throws IOException {
+ String maxTasks = args[0];
- Map props = new HashMap();
- props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
- props.put("offset.storage.file.filename", "/tmp/connect.offsets");
- // fast flushing for testing.
- props.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "10");
+ Map props = new HashMap();
+ props.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+ props.put("offset.storage.file.filename", "/tmp/connect.offsets");
+ // fast flushing for testing.
+ props.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "10");
- props.put("internal.key.converter.schemas.enable", "false");
- props.put("internal.value.converter.schemas.enable", "false");
- props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
- props.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
- props.put("key.converter.schemas.enable", "false");
- props.put("value.converter.schemas.enable", "false");
+ props.put("internal.key.converter.schemas.enable", "false");
+ props.put("internal.value.converter.schemas.enable", "false");
+ props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG,
+ "org.apache.kafka.connect.storage.StringConverter");
+ props.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG,
+ "org.apache.kafka.connect.storage.StringConverter");
+ props.put("key.converter.schemas.enable", "false");
+ props.put("value.converter.schemas.enable", "false");
- WorkerConfig workerCfg = new StandaloneConfig(props);
+ WorkerConfig workerCfg = new StandaloneConfig(props);
- MemoryOffsetBackingStore offBackingStore = new MemoryOffsetBackingStore();
- offBackingStore.configure(workerCfg);
+ MemoryOffsetBackingStore offBackingStore = new MemoryOffsetBackingStore();
+ offBackingStore.configure(workerCfg);
- Worker worker = new Worker("WORKER_ID", new SystemTime(), new Plugins(props), workerCfg, offBackingStore, new AllConnectorClientConfigOverridePolicy());
- worker.start();
+ Worker worker = new Worker("WORKER_ID", new SystemTime(), new Plugins(props), workerCfg,
+ offBackingStore, new AllConnectorClientConfigOverridePolicy());
+ worker.start();
- Herder herder = new StandaloneHerder(worker, ConnectUtils.lookupKafkaClusterId(workerCfg), new AllConnectorClientConfigOverridePolicy());
- herder.start();
+ Herder herder = new StandaloneHerder(worker, ConnectUtils.lookupKafkaClusterId(workerCfg),
+ new AllConnectorClientConfigOverridePolicy());
+ herder.start();
- Map<String, String> sourceProps = new HashMap<>();
- sourceProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSource.class.getName());
- sourceProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-source-connector");
- sourceProps.put(ConnectorConfig.TASKS_MAX_CONFIG, maxTasks);
- sourceProps.put(REGION_TO_TOPIC_BINDINGS, TEST_REGION_TO_TOPIC_BINDINGS);
+ Map<String, String> sourceProps = new HashMap<>();
+ sourceProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSource.class.getName());
+ sourceProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-source-connector");
+ sourceProps.put(ConnectorConfig.TASKS_MAX_CONFIG, maxTasks);
+ sourceProps.put(REGION_TO_TOPIC_BINDINGS, TEST_REGION_TO_TOPIC_BINDINGS);
- herder.putConnectorConfig(
- sourceProps.get(ConnectorConfig.NAME_CONFIG),
- sourceProps, true, (error, result)->{
- });
+ herder.putConnectorConfig(
+ sourceProps.get(ConnectorConfig.NAME_CONFIG),
+ sourceProps, true, (error, result) -> {
+ });
- Map<String, String> sinkProps = new HashMap<>();
- sinkProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSink.class.getName());
- sinkProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-sink-connector");
- sinkProps.put(ConnectorConfig.TASKS_MAX_CONFIG, maxTasks);
- sinkProps.put(TOPIC_TO_REGION_BINDINGS, TEST_TOPIC_TO_REGION_BINDINGS);
- sinkProps.put("topics", TEST_TOPIC_FOR_SINK);
+ Map<String, String> sinkProps = new HashMap<>();
+ sinkProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, GeodeKafkaSink.class.getName());
+ sinkProps.put(ConnectorConfig.NAME_CONFIG, "geode-kafka-sink-connector");
+ sinkProps.put(ConnectorConfig.TASKS_MAX_CONFIG, maxTasks);
+ sinkProps.put(TOPIC_TO_REGION_BINDINGS, TEST_TOPIC_TO_REGION_BINDINGS);
+ sinkProps.put("topics", TEST_TOPIC_FOR_SINK);
- herder.putConnectorConfig(
- sinkProps.get(ConnectorConfig.NAME_CONFIG),
- sinkProps, true, (error, result)->{
- });
+ herder.putConnectorConfig(
+ sinkProps.get(ConnectorConfig.NAME_CONFIG),
+ sinkProps, true, (error, result) -> {
+ });
- }
+ }
}
diff --git a/src/test/java/geode/kafka/ZooKeeperLocalCluster.java b/src/test/java/geode/kafka/ZooKeeperLocalCluster.java
index c1d743c..717b046 100644
--- a/src/test/java/geode/kafka/ZooKeeperLocalCluster.java
+++ b/src/test/java/geode/kafka/ZooKeeperLocalCluster.java
@@ -14,14 +14,14 @@
*/
package geode.kafka;
+import java.io.IOException;
+import java.util.Properties;
+
import org.apache.zookeeper.server.ServerConfig;
import org.apache.zookeeper.server.ZooKeeperServerMain;
import org.apache.zookeeper.server.admin.AdminServer;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
-import java.io.IOException;
-import java.util.Properties;
-
public class ZooKeeperLocalCluster {
ZooKeeperServerMain zooKeeperServer;
diff --git a/src/test/java/geode/kafka/sink/BatchRecordsTest.java b/src/test/java/geode/kafka/sink/BatchRecordsTest.java
index 210480e..cdb286b 100644
--- a/src/test/java/geode/kafka/sink/BatchRecordsTest.java
+++ b/src/test/java/geode/kafka/sink/BatchRecordsTest.java
@@ -14,84 +14,85 @@
*/
package geode.kafka.sink;
-import org.apache.geode.cache.Region;
-import org.apache.kafka.connect.sink.SinkRecord;
-import org.junit.Test;
-
-import java.util.Collection;
-import java.util.Map;
-
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.junit.Test;
+
+import org.apache.geode.cache.Region;
+
public class BatchRecordsTest {
- @Test
- public void updatingARecordShouldRemoveFromTheRemoveListIfNullValuesIsRemoveBooleanIsSet() {
- Map updates = mock(Map.class);
- Collection removes = mock(Collection.class);
- when(removes.contains(any())).thenReturn(true);
- BatchRecords records = new BatchRecords(updates, removes);
- SinkRecord sinkRecord = mock(SinkRecord.class);
- records.addUpdateOperation(sinkRecord, true);
- verify(removes, times(1)).remove(any());
- }
+ @Test
+ public void updatingARecordShouldRemoveFromTheRemoveListIfNullValuesIsRemoveBooleanIsSet() {
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(removes.contains(any())).thenReturn(true);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addUpdateOperation(sinkRecord, true);
+ verify(removes, times(1)).remove(any());
+ }
- @Test
- public void updatingARecordShouldAddToTheUpdateMap() {
- Map updates = mock(Map.class);
- Collection removes = mock(Collection.class);
- when(removes.contains(any())).thenReturn(false);
- BatchRecords records = new BatchRecords(updates, removes);
- SinkRecord sinkRecord = mock(SinkRecord.class);
- records.addUpdateOperation(sinkRecord, true);
- verify(updates, times(1)).put(any(), any());
- }
+ @Test
+ public void updatingARecordShouldAddToTheUpdateMap() {
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(removes.contains(any())).thenReturn(false);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addUpdateOperation(sinkRecord, true);
+ verify(updates, times(1)).put(any(), any());
+ }
- @Test
- public void updatingARecordShouldNotRemoveFromTheRemoveListIfNullValuesIsNotSet() {
- boolean nullValuesMeanRemove = false;
- Map updates = mock(Map.class);
- Collection removes = mock(Collection.class);
- when(removes.contains(any())).thenReturn(true);
- BatchRecords records = new BatchRecords(updates, removes);
- SinkRecord sinkRecord = mock(SinkRecord.class);
- records.addUpdateOperation(sinkRecord, nullValuesMeanRemove);
- verify(removes, times(0)).remove(any());
- }
+ @Test
+ public void updatingARecordShouldNotRemoveFromTheRemoveListIfNullValuesIsNotSet() {
+ boolean nullValuesMeanRemove = false;
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(removes.contains(any())).thenReturn(true);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addUpdateOperation(sinkRecord, nullValuesMeanRemove);
+ verify(removes, times(0)).remove(any());
+ }
- @Test
- public void removingARecordShouldRemoveFromTheUpdateMapIfKeyIsPresent() {
- Map updates = mock(Map.class);
- Collection removes = mock(Collection.class);
- when(updates.containsKey(any())).thenReturn(true);
- BatchRecords records = new BatchRecords(updates, removes);
- SinkRecord sinkRecord = mock(SinkRecord.class);
- records.addRemoveOperation(sinkRecord);
- verify(updates, times(1)).remove(any());
- }
+ @Test
+ public void removingARecordShouldRemoveFromTheUpdateMapIfKeyIsPresent() {
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ when(updates.containsKey(any())).thenReturn(true);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addRemoveOperation(sinkRecord);
+ verify(updates, times(1)).remove(any());
+ }
- @Test
- public void removingARecordAddToTheRemoveCollection() {
- Map updates = mock(Map.class);
- Collection removes = mock(Collection.class);
- BatchRecords records = new BatchRecords(updates, removes);
- SinkRecord sinkRecord = mock(SinkRecord.class);
- records.addRemoveOperation(sinkRecord);
- verify(removes, times(1)).add(any());
- }
+ @Test
+ public void removingARecordAddToTheRemoveCollection() {
+ Map updates = mock(Map.class);
+ Collection removes = mock(Collection.class);
+ BatchRecords records = new BatchRecords(updates, removes);
+ SinkRecord sinkRecord = mock(SinkRecord.class);
+ records.addRemoveOperation(sinkRecord);
+ verify(removes, times(1)).add(any());
+ }
- @Test
- public void executeOperationsShouldInvokePutAllAndRemoveAll() {
- Region region = mock(Region.class);
- BatchRecords records = new BatchRecords();
- records.executeOperations(region);
- verify(region, times(1)).putAll(any());
- verify(region, times(1)).removeAll(any());
- }
+ @Test
+ public void executeOperationsShouldInvokePutAllAndRemoveAll() {
+ Region region = mock(Region.class);
+ BatchRecords records = new BatchRecords();
+ records.executeOperations(region);
+ verify(region, times(1)).putAll(any());
+ verify(region, times(1)).removeAll(any());
+ }
diff --git a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
index ed8f040..735f255 100644
--- a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
+++ b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTaskTest.java
@@ -14,62 +14,61 @@
*/
package geode.kafka.sink;
-import geode.kafka.GeodeSinkConnectorConfig;
-import org.apache.geode.cache.Region;
-import org.apache.kafka.connect.sink.SinkRecord;
-import org.junit.Test;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-
import static geode.kafka.GeodeConnectorConfig.LOCATORS;
import static geode.kafka.GeodeConnectorConfig.TASK_ID;
import static geode.kafka.GeodeSinkConnectorConfig.NULL_VALUES_MEAN_REMOVE;
import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import geode.kafka.GeodeSinkConnectorConfig;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.junit.Test;
+
+import org.apache.geode.cache.Region;
+
public class GeodeKafkaSinkTaskTest {
- private HashMap<String, String> createTestSinkProps(boolean nullMeansRemove) {
- HashMap<String, String> props = new HashMap<>();
- props.put(TOPIC_TO_REGION_BINDINGS, "[topic:region]");
- props.put(TASK_ID, "0");
- props.put(NULL_VALUES_MEAN_REMOVE, String.valueOf(nullMeansRemove));
- props.put(LOCATORS, "localhost[10334]");
- return props;
- }
+ private HashMap<String, String> createTestSinkProps(boolean nullMeansRemove) {
+ HashMap<String, String> props = new HashMap<>();
+ props.put(TOPIC_TO_REGION_BINDINGS, "[topic:region]");
+ props.put(TASK_ID, "0");
+ props.put(NULL_VALUES_MEAN_REMOVE, String.valueOf(nullMeansRemove));
+ props.put(LOCATORS, "localhost[10334]");
+ return props;
+ }
- @Test
- public void putRecordsAddsToRegionBatchRecords() {
- boolean nullMeansRemove = true;
- GeodeKafkaSinkTask task = new GeodeKafkaSinkTask();
- HashMap<String, String> props = createTestSinkProps(nullMeansRemove);
+ @Test
+ public void putRecordsAddsToRegionBatchRecords() {
+ boolean nullMeansRemove = true;
+ GeodeKafkaSinkTask task = new GeodeKafkaSinkTask();
+ HashMap<String, String> props = createTestSinkProps(nullMeansRemove);
- SinkRecord topicRecord = mock(SinkRecord.class);
- when(topicRecord.topic()).thenReturn("topic");
- when(topicRecord.value()).thenReturn("value");
- when(topicRecord.key()).thenReturn("key");
+ SinkRecord topicRecord = mock(SinkRecord.class);
+ when(topicRecord.topic()).thenReturn("topic");
+ when(topicRecord.value()).thenReturn("value");
+ when(topicRecord.key()).thenReturn("key");
- List<SinkRecord> records = new ArrayList();
- records.add(topicRecord);
+ List<SinkRecord> records = new ArrayList();
+ records.add(topicRecord);
- HashMap<String, Region> regionNameToRegion = new HashMap<>();
- GeodeSinkConnectorConfig geodeSinkConnectorConfig = new GeodeSinkConnectorConfig(props);
- HashMap<String, BatchRecords> batchRecordsMap = new HashMap();
- BatchRecords batchRecords = mock(BatchRecords.class);
- batchRecordsMap.put("region", batchRecords);
- task.configure(geodeSinkConnectorConfig);
- task.setRegionNameToRegion(regionNameToRegion);
+ HashMap<String, Region> regionNameToRegion = new HashMap<>();
+ GeodeSinkConnectorConfig geodeSinkConnectorConfig = new GeodeSinkConnectorConfig(props);
+ HashMap<String, BatchRecords> batchRecordsMap = new HashMap();
+ BatchRecords batchRecords = mock(BatchRecords.class);
+ batchRecordsMap.put("region", batchRecords);
+ task.configure(geodeSinkConnectorConfig);
+ task.setRegionNameToRegion(regionNameToRegion);
- task.put(records, batchRecordsMap);
- assertTrue(batchRecordsMap.containsKey("region"));
- verify(batchRecords, times(1)).addUpdateOperation(topicRecord, nullMeansRemove);
- }
+ task.put(records, batchRecordsMap);
+ assertTrue(batchRecordsMap.containsKey("region"));
+ verify(batchRecords, times(1)).addUpdateOperation(topicRecord, nullMeansRemove);
+ }
}
diff --git a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java
index 633c7ae..d45c0c8 100644
--- a/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java
+++ b/src/test/java/geode/kafka/sink/GeodeKafkaSinkTest.java
@@ -14,52 +14,52 @@
*/
package geode.kafka.sink;
-import org.junit.Test;
+import static geode.kafka.GeodeConnectorConfig.TASK_ID;
+import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
-import static geode.kafka.GeodeConnectorConfig.TASK_ID;
-import static geode.kafka.GeodeSinkConnectorConfig.TOPIC_TO_REGION_BINDINGS;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import org.junit.Test;
public class GeodeKafkaSinkTest {
- @Test
- public void taskConfigsCreatesMaxNumberOfTasks() {
- GeodeKafkaSink sink = new GeodeKafkaSink();
- Map<String, String> props = new HashMap();
- props.put(TOPIC_TO_REGION_BINDINGS, "[someTopic:someRegion]");
- sink.start(props);
- Collection<Map<String,String>> tasks = sink.taskConfigs(5);
- assertEquals(5, tasks.size());
- }
+ @Test
+ public void taskConfigsCreatesMaxNumberOfTasks() {
+ GeodeKafkaSink sink = new GeodeKafkaSink();
+ Map<String, String> props = new HashMap();
+ props.put(TOPIC_TO_REGION_BINDINGS, "[someTopic:someRegion]");
+ sink.start(props);
+ Collection<Map<String, String>> tasks = sink.taskConfigs(5);
+ assertEquals(5, tasks.size());
+ }
- @Test
- public void sinkTaskConfigsAllAssignedEntireTopicToRegionBinding() {
- GeodeKafkaSink sink = new GeodeKafkaSink();
- Map<String, String> props = new HashMap();
- props.put(TOPIC_TO_REGION_BINDINGS, "[someTopic:someRegion]");
- sink.start(props);
- Collection<Map<String,String>> tasks = sink.taskConfigs(5);
- for(Map<String, String> prop : tasks) {
- assertEquals("[someTopic:someRegion]", prop.get(TOPIC_TO_REGION_BINDINGS));
- }
+ @Test
+ public void sinkTaskConfigsAllAssignedEntireTopicToRegionBinding() {
+ GeodeKafkaSink sink = new GeodeKafkaSink();
+ Map<String, String> props = new HashMap();
+ props.put(TOPIC_TO_REGION_BINDINGS, "[someTopic:someRegion]");
+ sink.start(props);
+ Collection<Map<String, String>> tasks = sink.taskConfigs(5);
+ for (Map<String, String> prop : tasks) {
+ assertEquals("[someTopic:someRegion]", prop.get(TOPIC_TO_REGION_BINDINGS));
}
+ }
- @Test
- public void eachTaskHasUniqueTaskIds() {
- GeodeKafkaSink sink = new GeodeKafkaSink();
- Map<String, String> props = new HashMap();
- props.put(TOPIC_TO_REGION_BINDINGS, "[someTopic:someRegion]");
- sink.start(props);
- Collection<Map<String,String>> tasks = sink.taskConfigs(5);
- HashSet<String> seenIds = new HashSet();
- for(Map<String, String> taskProp : tasks) {
- assertTrue(seenIds.add(taskProp.get(TASK_ID)));
- }
+ @Test
+ public void eachTaskHasUniqueTaskIds() {
+ GeodeKafkaSink sink = new GeodeKafkaSink();
+ Map<String, String> props = new HashMap();
+ props.put(TOPIC_TO_REGION_BINDINGS, "[someTopic:someRegion]");
+ sink.start(props);
+ Collection<Map<String, String>> tasks = sink.taskConfigs(5);
+ HashSet<String> seenIds = new HashSet();
+ for (Map<String, String> taskProp : tasks) {
+ assertTrue(seenIds.add(taskProp.get(TASK_ID)));
}
+ }
}
diff --git a/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java b/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
index 33f1ab5..a919b96 100644
--- a/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
+++ b/src/test/java/geode/kafka/source/GeodeKafkaSourceTaskTest.java
@@ -14,20 +14,6 @@
*/
package geode.kafka.source;
-import geode.kafka.GeodeContext;
-import org.apache.geode.cache.client.ClientCache;
-import org.apache.geode.cache.query.CqEvent;
-import org.junit.Test;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
import static geode.kafka.source.GeodeSourceConnectorConfig.DEFAULT_CQ_PREFIX;
import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_PARTITION;
import static org.hamcrest.CoreMatchers.is;
@@ -41,190 +27,212 @@
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import geode.kafka.GeodeContext;
+import org.junit.Test;
+
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.query.CqEvent;
+
public class GeodeKafkaSourceTaskTest {
- @Test
- public void whenLoadingEntireRegionAbleToPutInitialResultsIntoEventBuffer() {
- GeodeContext geodeContext = mock(GeodeContext.class);
- BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
- boolean loadEntireRegion = true;
- boolean isDurable = false;
- List<CqEvent> fakeInitialResults = new LinkedList<>();
- for (int i = 0; i < 10; i++) {
- fakeInitialResults.add(mock(CqEvent.class));
- }
-
- when(geodeContext.newCqWithInitialResults(anyString(), anyString(), any(), anyBoolean())).thenReturn(fakeInitialResults);
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- task.installListenersToRegion(geodeContext, 1, createEventBufferSupplier(eventBuffer), "testRegion", DEFAULT_CQ_PREFIX, loadEntireRegion, isDurable);
- assertEquals(10, eventBuffer.size());
+ @Test
+ public void whenLoadingEntireRegionAbleToPutInitialResultsIntoEventBuffer() {
+ GeodeContext geodeContext = mock(GeodeContext.class);
+ BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
+ boolean loadEntireRegion = true;
+ boolean isDurable = false;
+ List<CqEvent> fakeInitialResults = new LinkedList<>();
+ for (int i = 0; i < 10; i++) {
+ fakeInitialResults.add(mock(CqEvent.class));
}
- @Test
- public void whenNotLoadingEntireRegionShouldNotPutInitialResultsIntoEventBuffer() {
- GeodeContext geodeContext = mock(GeodeContext.class);
- BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
- boolean loadEntireRegion = false;
- boolean isDurable = false;
- List<CqEvent> fakeInitialResults = new LinkedList<>();
- for (int i = 0; i < 10; i++) {
- fakeInitialResults.add(mock(CqEvent.class));
- }
+ when(geodeContext.newCqWithInitialResults(anyString(), anyString(), any(), anyBoolean()))
+ .thenReturn(fakeInitialResults);
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ task.installListenersToRegion(geodeContext, 1, createEventBufferSupplier(eventBuffer),
+ "testRegion", DEFAULT_CQ_PREFIX, loadEntireRegion, isDurable);
+ assertEquals(10, eventBuffer.size());
+ }
- when(geodeContext.newCqWithInitialResults(anyString(), anyString(), any(), anyBoolean())).thenReturn(fakeInitialResults);
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- task.installListenersToRegion(geodeContext, 1, createEventBufferSupplier(eventBuffer), "testRegion", DEFAULT_CQ_PREFIX, loadEntireRegion, isDurable);
- assertEquals(0, eventBuffer.size());
+ @Test
+ public void whenNotLoadingEntireRegionShouldNotPutInitialResultsIntoEventBuffer() {
+ GeodeContext geodeContext = mock(GeodeContext.class);
+ BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
+ boolean loadEntireRegion = false;
+ boolean isDurable = false;
+ List<CqEvent> fakeInitialResults = new LinkedList<>();
+ for (int i = 0; i < 10; i++) {
+ fakeInitialResults.add(mock(CqEvent.class));
}
- @Test
- public void cqListenerOnEventPopulatesEventsBuffer() {
- GeodeContext geodeContext = mock(GeodeContext.class);
- BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
- boolean loadEntireRegion = false;
- boolean isDurable = false;
+ when(geodeContext.newCqWithInitialResults(anyString(), anyString(), any(), anyBoolean()))
+ .thenReturn(fakeInitialResults);
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ task.installListenersToRegion(geodeContext, 1, createEventBufferSupplier(eventBuffer),
+ "testRegion", DEFAULT_CQ_PREFIX, loadEntireRegion, isDurable);
+ assertEquals(0, eventBuffer.size());
+ }
- when(geodeContext.newCqWithInitialResults(anyString(), anyString(), any(), anyBoolean())).thenReturn(new ArrayList());
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- GeodeKafkaSourceListener listener = task.installListenersToRegion(geodeContext, 1, createEventBufferSupplier(eventBuffer), "testRegion", DEFAULT_CQ_PREFIX, loadEntireRegion, isDurable);
+ @Test
+ public void cqListenerOnEventPopulatesEventsBuffer() {
+ GeodeContext geodeContext = mock(GeodeContext.class);
+ BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
+ boolean loadEntireRegion = false;
+ boolean isDurable = false;
- listener.onEvent(mock(CqEvent.class));
- assertEquals(1, eventBuffer.size());
- }
+ when(geodeContext.newCqWithInitialResults(anyString(), anyString(), any(), anyBoolean()))
+ .thenReturn(new ArrayList());
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ GeodeKafkaSourceListener listener =
+ task.installListenersToRegion(geodeContext, 1, createEventBufferSupplier(eventBuffer),
+ "testRegion", DEFAULT_CQ_PREFIX, loadEntireRegion, isDurable);
- @Test
- public void readyForEventsIsCalledIfDurable() {
- ClientCache clientCache = mock(ClientCache.class);
+ listener.onEvent(mock(CqEvent.class));
+ assertEquals(1, eventBuffer.size());
+ }
- GeodeContext geodeContext = mock(GeodeContext.class);
- when(geodeContext.getClientCache()).thenReturn(clientCache);
+ @Test
+ public void readyForEventsIsCalledIfDurable() {
+ ClientCache clientCache = mock(ClientCache.class);
- GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
- when (config.isDurable()).thenReturn(true);
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- task.installOnGeode(config, geodeContext, null, "", false);
- verify(clientCache, times(1)).readyForEvents();
- }
+ GeodeContext geodeContext = mock(GeodeContext.class);
+ when(geodeContext.getClientCache()).thenReturn(clientCache);
- @Test
- public void cqIsInvokedForEveryRegionWithATopic() {
- ClientCache clientCache = mock(ClientCache.class);
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
+ when(config.isDurable()).thenReturn(true);
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ task.installOnGeode(config, geodeContext, null, "", false);
+ verify(clientCache, times(1)).readyForEvents();
+ }
- GeodeContext geodeContext = mock(GeodeContext.class);
- when(geodeContext.getClientCache()).thenReturn(clientCache);
+ @Test
+ public void cqIsInvokedForEveryRegionWithATopic() {
+ ClientCache clientCache = mock(ClientCache.class);
- Map<String, List<String>> regionToTopicsMap = new HashMap<>();
- regionToTopicsMap.put("region1", new ArrayList());
+ GeodeContext geodeContext = mock(GeodeContext.class);
+ when(geodeContext.getClientCache()).thenReturn(clientCache);
- GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
- when (config.getCqsToRegister()).thenReturn(regionToTopicsMap.keySet());
+ Map<String, List<String>> regionToTopicsMap = new HashMap<>();
+ regionToTopicsMap.put("region1", new ArrayList());
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- task.installOnGeode(config, geodeContext, null, "someCqPrefix", false);
- verify(geodeContext, times(1)).newCq(anyString(), anyString(), any(), anyBoolean());
- }
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
+ when(config.getCqsToRegister()).thenReturn(regionToTopicsMap.keySet());
- @Test
- public void cqWithInitialResultsIsInvokedForEveryRegionWithATopicIfLoadEntireIsSet() {
- ClientCache clientCache = mock(ClientCache.class);
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ task.installOnGeode(config, geodeContext, null, "someCqPrefix", false);
+ verify(geodeContext, times(1)).newCq(anyString(), anyString(), any(), anyBoolean());
+ }
- GeodeContext geodeContext = mock(GeodeContext.class);
- when(geodeContext.getClientCache()).thenReturn(clientCache);
+ @Test
+ public void cqWithInitialResultsIsInvokedForEveryRegionWithATopicIfLoadEntireIsSet() {
+ ClientCache clientCache = mock(ClientCache.class);
- Map<String, List<String>> regionToTopicsMap = new HashMap<>();
- regionToTopicsMap.put("region1", new ArrayList());
+ GeodeContext geodeContext = mock(GeodeContext.class);
+ when(geodeContext.getClientCache()).thenReturn(clientCache);
- GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
- when (config.getCqsToRegister()).thenReturn(regionToTopicsMap.keySet());
+ Map<String, List<String>> regionToTopicsMap = new HashMap<>();
+ regionToTopicsMap.put("region1", new ArrayList());
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- task.installOnGeode(config, geodeContext, createEventBufferSupplier(new LinkedBlockingQueue<>()), "someCqPrefix", true);
- verify(geodeContext, times(1)).newCqWithInitialResults(anyString(), anyString(), any(), anyBoolean());
- }
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
+ when(config.getCqsToRegister()).thenReturn(regionToTopicsMap.keySet());
- @Test
- public void readyForEventsIsNotCalledIfNotDurable() {
- ClientCache clientCache = mock(ClientCache.class);
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ task.installOnGeode(config, geodeContext,
+ createEventBufferSupplier(new LinkedBlockingQueue<>()), "someCqPrefix", true);
+ verify(geodeContext, times(1)).newCqWithInitialResults(anyString(), anyString(), any(),
+ anyBoolean());
+ }
- GeodeContext geodeContext = mock(GeodeContext.class);
- when(geodeContext.getClientCache()).thenReturn(clientCache);
+ @Test
+ public void readyForEventsIsNotCalledIfNotDurable() {
+ ClientCache clientCache = mock(ClientCache.class);
- GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
- when (config.isDurable()).thenReturn(false);
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- task.installOnGeode(config, geodeContext, null, "", false);
- verify(clientCache, times(0)).readyForEvents();
- }
+ GeodeContext geodeContext = mock(GeodeContext.class);
+ when(geodeContext.getClientCache()).thenReturn(clientCache);
- @Test
- public void pollReturnsEventsWhenEventBufferHasValues() throws Exception {
-// BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
-// CqEvent cqEvent = mock(CqEvent.class);
-// when(cqEvent.getNewValue()).thenReturn("New Value");
-// GeodeEvent event = mock(GeodeEvent.class);
-// when(event.getEvent()).thenReturn(cqEvent);
-// eventBuffer.add(event);
-//
-// List<String> topics = new ArrayList<>();
-// topics.add("myTopic");
-//
-// GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
-// task.startForTesting(eventBuffer, topics, 1);
-// List<SourceRecord> records = task.poll();
-// assertEquals(1, records.size());
- }
+ GeodeSourceConnectorConfig config = mock(GeodeSourceConnectorConfig.class);
+ when(config.isDurable()).thenReturn(false);
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ task.installOnGeode(config, geodeContext, null, "", false);
+ verify(clientCache, times(0)).readyForEvents();
+ }
- @Test
- public void installOnGeodeShouldCallCq() {
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- }
+ @Test
+ public void pollReturnsEventsWhenEventBufferHasValues() throws Exception {
+ // BlockingQueue<GeodeEvent> eventBuffer = new LinkedBlockingQueue(100);
+ // CqEvent cqEvent = mock(CqEvent.class);
+ // when(cqEvent.getNewValue()).thenReturn("New Value");
+ // GeodeEvent event = mock(GeodeEvent.class);
+ // when(event.getEvent()).thenReturn(cqEvent);
+ // eventBuffer.add(event);
+ //
+ // List<String> topics = new ArrayList<>();
+ // topics.add("myTopic");
+ //
+ // GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ // task.startForTesting(eventBuffer, topics, 1);
+ // List<SourceRecord> records = task.poll();
+ // assertEquals(1, records.size());
+ }
+
+ @Test
+ public void installOnGeodeShouldCallCq() {
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ }
+ @Test
+ public void createSourcePartitionsShouldReturnAMapOfSourcePartitions() {
+ GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ List<String> regionNames = Arrays.asList(new String[] {"region1", "region2", "region3"});
+ Map<String, Map<String, String>> sourcePartitions = task.createSourcePartitionsMap(regionNames);
+ assertThat(3, is(sourcePartitions.size()));
+ assertThat(true, is(sourcePartitions.get("region1").get(REGION_PARTITION).equals("region1")));
+ assertThat(true, is(sourcePartitions.get("region2").get(REGION_PARTITION).equals("region2")));
+ assertThat(true, is(sourcePartitions.get("region3").get(REGION_PARTITION).equals("region3")));
+ }
+
+ @Test
+ public void listOfLocatorsShouldBeConfiguredIntoClientCache() {
+
+ }
+
+ @Test
+ public void shouldNotBeDurableIfDurableClientIdIsNull() {
+
+ }
+
+ @Test
+ public void shouldNotCallReadyForEventsIfDurableClientPrefixIsEmpty() {
+
+ }
- @Test
- public void createSourcePartitionsShouldReturnAMapOfSourcePartitions() {
- GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- List<String> regionNames = Arrays.asList(new String[]{"region1", "region2", "region3"});
- Map<String, Map<String,String>> sourcePartitions = task.createSourcePartitionsMap(regionNames);
- assertThat(3, is(sourcePartitions.size()));
- assertThat(true, is(sourcePartitions.get("region1").get(REGION_PARTITION).equals("region1")));
- assertThat(true, is(sourcePartitions.get("region2").get(REGION_PARTITION).equals("region2")));
- assertThat(true, is(sourcePartitions.get("region3").get(REGION_PARTITION).equals("region3")));
- }
-
- @Test
- public void listOfLocatorsShouldBeConfiguredIntoClientCache() {
-
- }
-
- @Test
- public void shouldNotBeDurableIfDurableClientIdIsNull() {
-
- }
-
- @Test
- public void shouldNotCallReadyForEventsIfDurableClientPrefixIsEmpty() {
-
- }
+ @Test
+ public void cqPrefixShouldBeProperlyCalculatedFromProps() {
+ // GeodeContext geodeContext = mock(GeodeContext.class);
+ // GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
+ }
- @Test
- public void cqPrefixShouldBeProperlyCalculatedFromProps() {
-// GeodeContext geodeContext = mock(GeodeContext.class);
-// GeodeKafkaSourceTask task = new GeodeKafkaSourceTask();
- }
-
-
- private EventBufferSupplier createEventBufferSupplier(BlockingQueue<GeodeEvent> eventBuffer) {
- return new EventBufferSupplier() {
- @Override
- public BlockingQueue<GeodeEvent> get() {
- return eventBuffer;
- }
- };
- }
+ private EventBufferSupplier createEventBufferSupplier(BlockingQueue<GeodeEvent> eventBuffer) {
+ return new EventBufferSupplier() {
+ @Override
+ public BlockingQueue<GeodeEvent> get() {
+ return eventBuffer;
+ }
+ };
+ }
}
diff --git a/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java b/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
index 3e61c77..6632d75 100644
--- a/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
+++ b/src/test/java/geode/kafka/source/GeodeKafkaSourceTest.java
@@ -14,53 +14,53 @@
*/
package geode.kafka.source;
-import org.junit.Test;
+import static geode.kafka.GeodeConnectorConfig.TASK_ID;
+import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
-import static geode.kafka.GeodeConnectorConfig.TASK_ID;
-import static geode.kafka.source.GeodeSourceConnectorConfig.REGION_TO_TOPIC_BINDINGS;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import org.junit.Test;
public class GeodeKafkaSourceTest {
- @Test
- public void taskConfigsCreatesMaxNumberOfTasks() {
- GeodeKafkaSource source = new GeodeKafkaSource();
- Map<String, String> props = new HashMap();
- props.put(REGION_TO_TOPIC_BINDINGS, "[someRegion:someTopic]");
- source.start(props);
- Collection<Map<String,String>> tasks = source.taskConfigs(5);
- assertEquals(5, tasks.size());
- }
+ @Test
+ public void taskConfigsCreatesMaxNumberOfTasks() {
+ GeodeKafkaSource source = new GeodeKafkaSource();
+ Map<String, String> props = new HashMap();
+ props.put(REGION_TO_TOPIC_BINDINGS, "[someRegion:someTopic]");
+ source.start(props);
+ Collection<Map<String, String>> tasks = source.taskConfigs(5);
+ assertEquals(5, tasks.size());
+ }
- @Test
- public void sourceTaskConfigsAllAssignedEntireRegionToTopicBinding() {
- GeodeKafkaSource source = new GeodeKafkaSource();
- Map<String, String> props = new HashMap();
- props.put(REGION_TO_TOPIC_BINDINGS, "[someRegion:someTopic]");
- source.start(props);
- Collection<Map<String,String>> tasks = source.taskConfigs(5);
- for(Map<String, String> prop : tasks) {
- assertEquals("[someRegion:someTopic]", prop.get(REGION_TO_TOPIC_BINDINGS));
- }
+ @Test
+ public void sourceTaskConfigsAllAssignedEntireRegionToTopicBinding() {
+ GeodeKafkaSource source = new GeodeKafkaSource();
+ Map<String, String> props = new HashMap();
+ props.put(REGION_TO_TOPIC_BINDINGS, "[someRegion:someTopic]");
+ source.start(props);
+ Collection<Map<String, String>> tasks = source.taskConfigs(5);
+ for (Map<String, String> prop : tasks) {
+ assertEquals("[someRegion:someTopic]", prop.get(REGION_TO_TOPIC_BINDINGS));
}
+ }
- @Test
- public void eachTaskHasUniqueTaskIds() {
- GeodeKafkaSource sink = new GeodeKafkaSource();
- Map<String, String> props = new HashMap();
- props.put(REGION_TO_TOPIC_BINDINGS, "[someRegion:someTopic]");
- sink.start(props);
- Collection<Map<String,String>> tasks = sink.taskConfigs(5);
- HashSet<String> seenIds = new HashSet();
- for(Map<String, String> taskProp : tasks) {
- assertTrue(seenIds.add(taskProp.get(TASK_ID)));
- }
+ @Test
+ public void eachTaskHasUniqueTaskIds() {
+ GeodeKafkaSource sink = new GeodeKafkaSource();
+ Map<String, String> props = new HashMap();
+ props.put(REGION_TO_TOPIC_BINDINGS, "[someRegion:someTopic]");
+ sink.start(props);
+ Collection<Map<String, String>> tasks = sink.taskConfigs(5);
+ HashSet<String> seenIds = new HashSet();
+ for (Map<String, String> taskProp : tasks) {
+ assertTrue(seenIds.add(taskProp.get(TASK_ID)));
}
+ }
}
diff --git a/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java b/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java
index c75e1b8..a545a72 100644
--- a/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java
+++ b/src/test/java/geode/kafka/source/GeodeSourceConnectorConfigTest.java
@@ -14,26 +14,26 @@
*/
package geode.kafka.source;
-import org.junit.Test;
-
-import java.util.HashMap;
-import java.util.Map;
-
import static geode.kafka.GeodeConnectorConfig.LOCATORS;
import static geode.kafka.GeodeConnectorConfig.TASK_ID;
import static geode.kafka.source.GeodeSourceConnectorConfig.DURABLE_CLIENT_ID_PREFIX;
import static org.junit.Assert.assertEquals;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
public class GeodeSourceConnectorConfigTest {
- @Test
- public void durableClientIdShouldNotBeSetIfPrefixIsEmpty() {
- Map<String, String> props = new HashMap<>();
- props.put(TASK_ID, "0");
- props.put(DURABLE_CLIENT_ID_PREFIX, "");
- props.put(LOCATORS, "localhost[10334]");
- GeodeSourceConnectorConfig config = new GeodeSourceConnectorConfig(props);
- assertEquals("", config.getDurableClientId());
- }
+ @Test
+ public void durableClientIdShouldNotBeSetIfPrefixIsEmpty() {
+ Map<String, String> props = new HashMap<>();
+ props.put(TASK_ID, "0");
+ props.put(DURABLE_CLIENT_ID_PREFIX, "");
+ props.put(LOCATORS, "localhost[10334]");
+ GeodeSourceConnectorConfig config = new GeodeSourceConnectorConfig(props);
+ assertEquals("", config.getDurableClientId());
+ }
}
diff --git a/src/test/java/geode/kafka/source/SharedEventBufferSupplierTest.java b/src/test/java/geode/kafka/source/SharedEventBufferSupplierTest.java
index b4a429b..87a0e07 100644
--- a/src/test/java/geode/kafka/source/SharedEventBufferSupplierTest.java
+++ b/src/test/java/geode/kafka/source/SharedEventBufferSupplierTest.java
@@ -1,51 +1,65 @@
+/*
+ * 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 geode.kafka.source;
-import org.junit.Test;
-
-import java.util.concurrent.BlockingQueue;
-
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.mockito.Mockito.mock;
+import java.util.concurrent.BlockingQueue;
+
+import org.junit.Test;
+
public class SharedEventBufferSupplierTest {
- @Test
- public void creatingNewSharedEventSupplierShouldCreateInstance() {
- SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
- assertNotNull(supplier.get());
- }
+ @Test
+ public void creatingNewSharedEventSupplierShouldCreateInstance() {
+ SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
+ assertNotNull(supplier.get());
+ }
- @Test
- public void alreadySharedEventSupplierShouldReturnSameInstanceOfEventBuffer() {
- SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
- BlockingQueue<GeodeEvent> queue = supplier.get();
- supplier = new SharedEventBufferSupplier(1);
- assertEquals(queue, supplier.get());
- }
+ @Test
+ public void alreadySharedEventSupplierShouldReturnSameInstanceOfEventBuffer() {
+ SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
+ BlockingQueue<GeodeEvent> queue = supplier.get();
+ supplier = new SharedEventBufferSupplier(1);
+ assertEquals(queue, supplier.get());
+ }
- @Test
- public void newEventBufferShouldBeReflectedInAllSharedSuppliers() {
- SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
- SharedEventBufferSupplier newSupplier = new SharedEventBufferSupplier(2);
- assertEquals(supplier.get(), newSupplier.get());
- }
+ @Test
+ public void newEventBufferShouldBeReflectedInAllSharedSuppliers() {
+ SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
+ SharedEventBufferSupplier newSupplier = new SharedEventBufferSupplier(2);
+ assertEquals(supplier.get(), newSupplier.get());
+ }
- @Test
- public void newEventBufferSuppliedShouldNotBeTheOldQueue() {
- SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
- BlockingQueue<GeodeEvent> queue = supplier.get();
- SharedEventBufferSupplier newSupplier = new SharedEventBufferSupplier(2);
- assertNotEquals(queue, newSupplier.get());
- }
+ @Test
+ public void newEventBufferSuppliedShouldNotBeTheOldQueue() {
+ SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
+ BlockingQueue<GeodeEvent> queue = supplier.get();
+ SharedEventBufferSupplier newSupplier = new SharedEventBufferSupplier(2);
+ assertNotEquals(queue, newSupplier.get());
+ }
- @Test
- public void newEventBufferShouldContainAllEventsFromTheOldSupplier() {
- SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
- GeodeEvent geodeEvent = mock(GeodeEvent.class);
- supplier.get().add(geodeEvent);
- SharedEventBufferSupplier newSupplier = new SharedEventBufferSupplier(2);
- assertEquals(geodeEvent, newSupplier.get().poll());
- }
+ @Test
+ public void newEventBufferShouldContainAllEventsFromTheOldSupplier() {
+ SharedEventBufferSupplier supplier = new SharedEventBufferSupplier(1);
+ GeodeEvent geodeEvent = mock(GeodeEvent.class);
+ supplier.get().add(geodeEvent);
+ SharedEventBufferSupplier newSupplier = new SharedEventBufferSupplier(2);
+ assertEquals(geodeEvent, newSupplier.get().poll());
+ }
}
