Merge latest commits from apache again

bin/copycat-distributed.sh

@@ -0,0 +1,23 @@
+#!/bin/sh
+# 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.
+
+base_dir=$(dirname $0)
+
+if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
+    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/copycat-log4j.properties"
+fi
+
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.copycat.cli.CopycatDistributed "$@"

checkstyle/import-control.xml

@@ -27,6 +27,9 @@
 	<allow pkg="javax.management" />
 	<allow pkg="org.slf4j" />
 	<allow pkg="org.junit" />
+	<allow pkg="org.easymock" />
+	<allow pkg="org.powermock" />
+
 	<allow pkg="javax.net.ssl" />
 
 	<!-- no one depends on the server -->
@@ -138,37 +141,33 @@
 		<allow pkg="org.apache.kafka.common" />
 		<allow pkg="org.apache.kafka.copycat.data" />
 		<allow pkg="org.apache.kafka.copycat.errors" />
+		<allow pkg="org.apache.kafka.clients" />
 
 		<subpackage name="source">
 			<allow pkg="org.apache.kafka.copycat.connector" />
 			<allow pkg="org.apache.kafka.copycat.storage" />
 		</subpackage>
 
 		<subpackage name="sink">
+		        <allow pkg="org.apache.kafka.clients.consumer" />
 			<allow pkg="org.apache.kafka.copycat.connector" />
 			<allow pkg="org.apache.kafka.copycat.storage" />
 		</subpackage>
 
 		<subpackage name="runtime">
 			<allow pkg="org.apache.kafka.copycat" />
-			<allow pkg="org.apache.kafka.clients" />
-			<!-- for tests -->
-			<allow pkg="org.easymock" />
-			<allow pkg="org.powermock" />
 		</subpackage>
 
 		<subpackage name="cli">
 			<allow pkg="org.apache.kafka.copycat.runtime" />
+			<allow pkg="org.apache.kafka.copycat.storage" />
 			<allow pkg="org.apache.kafka.copycat.util" />
 			<allow pkg="org.apache.kafka.common" />
 		</subpackage>
 
 		<subpackage name="storage">
 			<allow pkg="org.apache.kafka.copycat" />
 			<allow pkg="org.apache.kafka.common.serialization" />
-			<!-- for tests -->
-			<allow pkg="org.easymock" />
-			<allow pkg="org.powermock" />
 		</subpackage>
 
 		<subpackage name="util">
@@ -184,6 +183,7 @@
 
 		<subpackage name="file">
 			<allow pkg="org.apache.kafka.copycat" />
+		        <allow pkg="org.apache.kafka.clients.consumer" />
 			<!-- for tests -->
 			<allow pkg="org.easymock" />
 			<allow pkg="org.powermock" />

clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java

@@ -79,7 +79,7 @@ public interface Consumer<K, V> extends Closeable {
     /**
      * @see KafkaConsumer#commitSync(Map)
      */
-    public void commitSync(Map<TopicPartition, Long> offsets);
+    public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);
 
     /**
      * @see KafkaConsumer#commitAsync()
@@ -94,7 +94,7 @@ public interface Consumer<K, V> extends Closeable {
     /**
      * @see KafkaConsumer#commitAsync(Map, OffsetCommitCallback)
      */
-    public void commitAsync(Map<TopicPartition, Long> offsets, OffsetCommitCallback callback);
+    public void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);
 
     /**
      * @see KafkaConsumer#seek(TopicPartition, long)
@@ -119,7 +119,7 @@ public interface Consumer<K, V> extends Closeable {
     /**
      * @see KafkaConsumer#committed(TopicPartition)
      */
-    public long committed(TopicPartition partition);
+    public OffsetAndMetadata committed(TopicPartition partition);
 
     /**
      * @see KafkaConsumer#metrics()

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -48,6 +48,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.ConcurrentModificationException;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -776,6 +777,8 @@ public void assign(List<TopicPartition> partitions) {
      *
      * @throws NoOffsetForPartitionException If there is no stored offset for a subscribed partition and no automatic
      *             offset reset policy has been configured.
+     * @throws org.apache.kafka.common.errors.OffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse and
+     *         the defaultResetPolicy is NONE
      */
     @Override
     public ConsumerRecords<K, V> poll(long timeout) {
@@ -813,6 +816,8 @@ public ConsumerRecords<K, V> poll(long timeout) {
      * heart-beating, auto-commits, and offset updates.
      * @param timeout The maximum time to block in the underlying poll
      * @return The fetched records (may be empty)
+     * @throws org.apache.kafka.common.errors.OffsetOutOfRangeException If there is OffsetOutOfRange error in fetchResponse and
+     *         the defaultResetPolicy is NONE
      */
     private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollOnce(long timeout) {
         // TODO: Sub-requests should take into account the poll timeout (KAFKA-1894)
@@ -829,6 +834,12 @@ private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollOnce(long timeout) {
 
         // init any new fetches (won't resend pending fetches)
         Cluster cluster = this.metadata.fetch();
+        Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchedRecords();
+        // Avoid block waiting for response if we already have data available, e.g. from another API call to commit.
+        if (!records.isEmpty()) {
+            client.poll(0);
+            return records;
+        }
         fetcher.initFetches(cluster);
         client.poll(timeout);
         return fetcher.fetchedRecords();
@@ -839,7 +850,7 @@ private void scheduleAutoCommitTask(final long interval) {
             public void run(long now) {
                 commitAsync(new OffsetCommitCallback() {
                     @Override
-                    public void onComplete(Map<TopicPartition, Long> offsets, Exception exception) {
+                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                         if (exception != null)
                             log.error("Auto offset commit failed.", exception);
                     }
@@ -880,10 +891,10 @@ public void commitSync() {
      * This is a synchronous commits and will block until either the commit succeeds or an unrecoverable error is
      * encountered (in which case it is thrown to the caller).
      *
-     * @param offsets The list of offsets per partition that should be committed to Kafka.
+     * @param offsets A map of offsets by partition with associated metadata
      */
     @Override
-    public void commitSync(final Map<TopicPartition, Long> offsets) {
+    public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets) {
         acquire();
         try {
             coordinator.commitOffsetsSync(offsets);
@@ -932,15 +943,16 @@ public void commitAsync(OffsetCommitCallback callback) {
      * This is an asynchronous call and will not block. Any errors encountered are either passed to the callback
      * (if provided) or discarded.
      *
-     * @param offsets The list of offsets per partition that should be committed to Kafka.
+     * @param offsets A map of offsets by partition with associate metadata. This map will be copied internally, so it
+     *                is safe to mutate the map after returning.
      * @param callback Callback to invoke when the commit completes
      */
     @Override
-    public void commitAsync(final Map<TopicPartition, Long> offsets, OffsetCommitCallback callback) {
+    public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
         acquire();
         try {
             log.debug("Committing offsets: {} ", offsets);
-            coordinator.commitOffsetsAsync(offsets, callback);
+            coordinator.commitOffsetsAsync(new HashMap<>(offsets), callback);
         } finally {
             release();
         }
@@ -1013,10 +1025,9 @@ public long position(TopicPartition partition) {
             Long offset = this.subscriptions.consumed(partition);
             if (offset == null) {
                 updateFetchPositions(Collections.singleton(partition));
-                return this.subscriptions.consumed(partition);
-            } else {
-                return offset;
+                offset = this.subscriptions.consumed(partition);
             }
+            return offset;
         } finally {
             release();
         }
@@ -1030,29 +1041,24 @@ public long position(TopicPartition partition) {
      * consumer hasn't yet initialized its cache of committed offsets.
      *
      * @param partition The partition to check
-     * @return The last committed offset
-     * @throws NoOffsetForPartitionException If no offset has ever been committed by any process for the given
-     *             partition.
+     * @return The last committed offset and metadata or null if there was no prior commit
      */
     @Override
-    public long committed(TopicPartition partition) {
+    public OffsetAndMetadata committed(TopicPartition partition) {
         acquire();
         try {
-            Long committed;
+            OffsetAndMetadata committed;
             if (subscriptions.isAssigned(partition)) {
                 committed = this.subscriptions.committed(partition);
                 if (committed == null) {
                     coordinator.refreshCommittedOffsetsIfNeeded();
                     committed = this.subscriptions.committed(partition);
                 }
             } else {
-                Map<TopicPartition, Long> offsets = coordinator.fetchCommittedOffsets(Collections.singleton(partition));
+                Map<TopicPartition, OffsetAndMetadata> offsets = coordinator.fetchCommittedOffsets(Collections.singleton(partition));
                 committed = offsets.get(partition);
             }
 
-            if (committed == null)
-                throw new NoOffsetForPartitionException("No offset has been committed for partition " + partition);
-
             return committed;
         } finally {
             release();